Trigger refresh of Redash queries based on query tag

quick_date_ranges.py

from datetime import datetime, timedelta
from collections import namedtuple


def get_frontend_vals():

	ranges = calculate_ranges()
	singles = calculate_singletons()

	valkeys = [k for k in ranges.keys()] + [k for k in singles.keys()]

	Values = namedtuple('Values', ' '.join(valkeys))

	return Values(**ranges, **singles)

def calculate_ranges():

	SuperToday = namedtuple('SuperToday', 'month day year weeknum weekday')
	DateRange = namedtuple('DateRange', 'start end')

	today = datetime.today()
	_ymd = (today.month, today.day)

	t = SuperToday(*_ymd, *today.isocalendar())

	ranges = {}

	#  _____ _     _      __        __        _     
	# |_   _| |__ (_)___  \ \      / /__  ___| | __ 
	#   | | | '_ \| / __|  \ \ /\ / / _ \/ _ \ |/ / 
	#   | | | | | | \__ \   \ V  V /  __/  __/   <  
	#   |_| |_| |_|_|___/    \_/\_/ \___|\___|_|\_\ 

	start = datetime.strptime(f"{t.year}-{t.weeknum}-1", '%G-%V-%u')
	end = start + timedelta(days=6)

	ranges['d_this_week'] = DateRange(start,end)

											
	#  _____ _     _       __  __             _   _      
	# |_   _| |__ (_)___  |  \/  | ___  _ __ | |_| |__   
	#   | | | '_ \| / __| | |\/| |/ _ \| '_ \| __| '_ \  
	#   | | | | | | \__ \ | |  | | (_) | | | | |_| | | | 
	#   |_| |_| |_|_|___/ |_|  |_|\___/|_| |_|\__|_| |_| 

	start = datetime.strptime(f"{t.year}-{t.month}-1", '%Y-%m-%d')
	e_year, e_month = (t.year,t.month+1) if t.month < 12 else (t.year+1,1)
	end = datetime.strptime(
		f"{e_year}-{e_month}-1", '%Y-%m-%d') \
		- timedelta(days=1)

	ranges['d_this_month'] = DateRange(start,end)
											  
	#  _____ _     _      __   __                    
	# |_   _| |__ (_)___  \ \ / /__  __ _ _ __       
	#   | | | '_ \| / __|  \ V / _ \/ _` | '__|      
	#   | | | | | | \__ \   | |  __/ (_| | |         
	#   |_| |_| |_|_|___/   |_|\___|\__,_|_|         

	start = datetime.strptime(f"{t.year}-1-1", '%Y-%m-%d')
	end = datetime.strptime(f"{t.year}-12-31", '%Y-%m-%d')

	ranges['d_this_year'] = DateRange(start,end)

												  
	#  _              _    __        __        _       
	# | |    __ _ ___| |_  \ \      / /__  ___| | __   
	# | |   / _` / __| __|  \ \ /\ / / _ \/ _ \ |/ /   
	# | |__| (_| \__ \ |_    \ V  V /  __/  __/   <    
	# |_____\__,_|___/\__|    \_/\_/ \___|\___|_|\_\   

	start = datetime.strptime(f"{t.year}-{t.weeknum-1}-1", '%G-%V-%u')
	end = start + timedelta(days=6)

	ranges['d_last_week'] = DateRange(start,end)


	#  _              _     __  __             _   _      
	# | |    __ _ ___| |_  |  \/  | ___  _ __ | |_| |__   
	# | |   / _` / __| __| | |\/| |/ _ \| '_ \| __| '_ \  
	# | |__| (_| \__ \ |_  | |  | | (_) | | | | |_| | | | 
	# |_____\__,_|___/\__| |_|  |_|\___/|_| |_|\__|_| |_| 


	s_year, s_month = (t.year-1, 12) if t.month == 1 else (t.year,t.month-1)

	start = datetime.strptime(f"{s_year}-{s_month}-1", '%Y-%m-%d')
	end = datetime.strptime(f"{t.year}-{t.month}-1", '%Y-%m-%d') \
		- timedelta(days=1)

	ranges['d_last_month'] = DateRange(start, end)

	#  _              _    __   __                   
	# | |    __ _ ___| |_  \ \ / /__  __ _ _ __      
	# | |   / _` / __| __|  \ V / _ \/ _` | '__|     
	# | |__| (_| \__ \ |_    | |  __/ (_| | |        
	# |_____\__,_|___/\__|   |_|\___|\__,_|_|        

	start = datetime.strptime(f"{t.year-1}-1-1", '%Y-%m-%d')
	end = datetime.strptime(f"{t.year-1}-12-31", '%Y-%m-%d')

	ranges['d_last_year'] = DateRange(start,end)


	 #  _              _    __  __  ____                   
	 # | |    __ _ ___| |_  \ \/ / |  _ \  __ _ _   _ ___  
	 # | |   / _` / __| __|  \  /  | | | |/ _` | | | / __| 
	 # | |__| (_| \__ \ |_   /  \  | |_| | (_| | |_| \__ \ 
	 # |_____\__,_|___/\__| /_/\_\ |____/ \__,_|\__, |___/ 
	 #                                          |___/      

	def make_x_days_date_range(today, days):
		start = today - timedelta(days=days)
		end = today

		return DateRange(start, end)

	for x in [7, 14, 30, 60, 90]:
		ranges[f"d_last_{x}_days"] = make_x_days_date_range(today, x)

	return ranges

def calculate_singletons():

	today = datetime.today()
	d_now=datetime.strptime(
		f"{today.year}-{today.month}-{today.day}", '%Y-%m-%d')
	d_yesterday = d_now - timedelta(days=1)

	return dict(d_now=d_now, d_yesterday=d_yesterday)

trigger_refresh.py

from urllib.parse import urljoin
import requests
import click
import json

import quick_date_ranges as qdr


class Redash:
    def __init__(self, host, api_key):
        self.host = host
        self.api_key = api_key

    def _request(self, method, path, *args, **kwargs):
        if self.host.endswith("/"):
            url = self.host[0:-1] + path
        else:
            url = self.host + path

        kwargs["headers"] = {"Authorization": "Key {}".format(self.api_key)}
        response = requests.request(method, url, *args, **kwargs)

        return response.json()

    def get_queries(self, tag, skip_scheduled):
        if tag:
            params = {"tags": tag}
        else:
            params = {}

        stop_loading = False
        page = 1
        page_size = 100

        items = []

        while not stop_loading:
            params["page"] = page
            params["page_size"] = page_size
            response = self._request("get", "/api/queries", params=params)

            items += response["results"]
            page += 1

            stop_loading = response["page"] * response["page_size"] >= response["count"]

        if skip_scheduled:
            items = [query for query in items if query['schedule'] is None]

        return items

    def trigger_refresh(self, query, parameters):
        print("Triggering refresh of {}...".format(query['id']))

        data = {'max_age': 0, 'parameters': parameters}
        response = self._request('post', '/api/queries/{}/results'.format(query['id']), json=data)

        if 'job' in response:
            if 'id' in response['job']:
                print("Job id: {}".format(response['job']['id']))
            else:
                print("Error: {}".format(response['job']['error']))
        else:
            print(response)

        return response


@click.command()
@click.argument("redash_host")
@click.option(
    "--api-key",
    "api_key",
    envvar="REDASH_API_KEY",
    show_envvar=True,
    prompt="API Key",
    help="User API Key",
)
@click.option('--parameters', help="Query Parameters in JSON form", default='{}')
@click.option('--skip-scheduled', 'skip_scheduled', is_flag=True, default=False, help="Skip scheduled queries")
@click.option("--tag", default=None, help="Tag to filter the queries list with.")
def trigger(redash_host, api_key, tag, skip_scheduled, parameters):
    """Trigger query refresh on REDASH_HOST using given api-key and optional list of tags."""

    message = "Refreshing queries from {}"
    if tag:
        message += " with tag: {}"

    if skip_scheduled:
        message += " (skipping scheduled ones)"

    message += "..."

    print(message.format(redash_host, tag))

    redash = Redash(redash_host, api_key)
    queries = redash.get_queries(tag, skip_scheduled=skip_scheduled)

    print("Found {} queries to refresh.".format(len(queries)))

    # Dates and Times
    dat = qdr.get_frontend_vals()

    auto_params = {}

    for query in queries:

        for p in query['options']['parameters']:
            if not p['value'] in dat._fields:
                continue

            if p['value'] in ('d_now', 'd_yesterday'):
                str_date = getattr(dat, p['value']).isoformat()
                auto_params.setdefault(p['name'], str_date )
            else:
                str_start = getattr(dat, p['value']).start.isoformat()
                str_end = getattr(dat, p['value']).start.isoformat()
                auto_params.setdefault(p['name'],
                    dict(start=str_start, end=str_end))

        job = redash.trigger_refresh(query, auto_params)

if __name__ == "__main__":
    trigger()