influx_queries.py

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"
    )