from dataclasses import dataclass, field from itertools import repeat from typing import List, Union @dataclass class Query: select_: str from_: str where_: str group_by: List[str] = field(default_factory=list) def __str__(self): ret = f'SELECT \"{self.select_}\" ' ret += f'FROM \"{self.from_}\" ' ret += f'WHERE ({self.where_}) AND $timeFilter ' group_by = ', '.join(f'"{tag}"' for tag in self.group_by) ret += f'GROUP BY {group_by}' return ret def show_tag_values(table: str, key_name: str) -> str: """Return influx query to get all tag values from a measurment.""" return f"SHOW TAG VALUES FROM \"{table}\" WITH key = \"{key_name}\"" def get_variable_condition(variable_name: str) -> str: clean = variable_name.strip() if not clean: raise ValueError("Empty variable name") return f'"{clean}" =~ /^${clean}$/' def join_conditions(conditions: List[str], operators: Union[List[str], str]): ops = operators if isinstance(operators, str): ops = repeat(operators, len(conditions) - 1) elif len(operators) == 1: ops = repeat(operators[0], len(conditions) - 1) else: if len(conditions) - 1 != len(operators): raise ValueError("unfitting lengths of conditions and operators") ret = conditions[0] for op, cond in zip(ops, conditions[1:]): ret += f' {op} {cond}' return ret def join_variable_and(variable_names: List[str]) -> str: return join_conditions( [get_variable_condition(name) for name in variable_names], "AND" )