Reisacher-Forderung/app/forderung_data.py

"""One authorized projection for list results and dashboard aggregates."""

from sqlalchemy import text

from .jinja_templates import templates
from .query_filter import EMPTY_FILTER_LABEL, EMPTY_FILTER_VALUE


def read_filter_suggestions(db, username, field, clauses, values, search="", *, selected=None):
    # Reuse the authorized projection, including local changes to its WHERE clause.
    query = templates.env.get_template("forderungen/liste/queries/forderungen_liste.sql").render()
    query = query.rsplit("ORDER BY", 1)[0]
    raw_key, raw_label, sort = (f"F.[{column}]" for column in (field.key_column, field.value_column, field.sort_column))
    key = f"COALESCE(NULLIF(CAST({raw_key} AS nvarchar(4000)), ''), '{EMPTY_FILTER_VALUE}')"
    label = (
        f"CASE WHEN NULLIF(CAST({raw_key} AS nvarchar(4000)), '') IS NULL THEN '{EMPTY_FILTER_LABEL}' "
        f"ELSE CAST({raw_label} AS nvarchar(4000)) END"
    )
    limit = "TOP 51 " if selected is None else ""
    query = query.replace(
        "SELECT F.*", f"SELECT {limit}{key} AS value, MIN({label}) AS label, MIN({sort}) AS sort_value", 1
    )
    predicates = list(clauses)
    values = dict(values, auth_username=username)
    if selected is not None:
        placeholders = []
        for index, value in enumerate(selected):
            parameter = f"selected_{index}"
            placeholders.append(":" + parameter)
            values[parameter] = value
        if not placeholders:
            return []
        predicates.append(f"{key} IN ({', '.join(placeholders)})")
    if search:
        predicates.append(
            f"(CAST({key} AS nvarchar(max)) LIKE :suggest_search OR CAST({label} AS nvarchar(max)) LIKE :suggest_search)"
        )
        values["suggest_search"] = f"%{search}%"
    if predicates:
        query = query.replace("AND 1 = 1", "AND 1 = 1 AND " + " AND ".join(predicates))
    query += f" GROUP BY {key} ORDER BY MIN({sort}), {key}"
    rows = db.execute(text(query), values).fetchall()
    return [{"value": str(row[0]), "label": str(row[1] if row[1] is not None else row[0])} for row in rows]


def read_invoices(db, username, clauses=(), values=None):
    query = templates.env.get_template("forderungen/liste/queries/forderungen_liste.sql").render()
    if clauses:
        query = query.replace("AND 1 = 1", "AND 1 = 1 AND " + " AND ".join(clauses))
    result = db.execute(
        text(query).execution_options(query_name="forderungen/liste/queries/forderungen_liste.sql"),
        dict(values or {}, auth_username=username),
    )
    columns = list(result.keys())
    tenant, number = columns.index("Client_DB"), columns.index("Beleg_Nr")
    seen, rows = set(), []
    for row in result.fetchall():
        identity = (str(row[tenant]), str(row[number]))
        if identity not in seen:
            rows.append(row)
            seen.add(identity)
    return columns, rows