ServerKit/backend/app/api/_query.py

"""Shared list-query layer: ``$select`` / ``$filter`` / ``$orderby`` / ``$skip`` / ``$top``.

Ported from the author's ``conuco_v2`` query layer, with two deliberate changes.

**ORM, not Core.** conuco resolves columns through ``table.c``; ServerKit is
Flask-SQLAlchemy ORM, so fields resolve against the mapper's column set. That
set is also the allowlist — a field that is not a real mapped column is a 400,
never an ``getattr`` into arbitrary model attributes.

**``$select`` is not only a column narrowing.** This is where the actual win is
here. ``Application.to_dict()`` costs roughly three extra queries per row on top
of the row itself: ``image_scans.first()`` and ``image_update_checks.first()``
are ``lazy='dynamic'`` relationships (immune to eager loading — a dynamic
attribute *is* a Query), and ``sleep_policy`` / ``server`` / ``project`` /
``environment`` are lazy scalars. Drawing a 100-app list is ~300 queries. So
``parse_select`` hands back a *field set* that ``to_dict`` uses to skip the
loads it was not asked for: ``$select=id,name`` goes from 3 queries per row to
zero extra ones.

Not implemented here: ``$view``. conuco resolves a named server-side view into
default ``$filter``/``$orderby``/``$select``, but ServerKit's saved views are
per-user rows already resolved in the browser (see
``components/ds/grid/viewState.js``) — the grid knows the concrete config and
can serialize it, so a second server-side copy of the same concept would be a
place for the two to disagree.

Usage::

    from app.api._query import apply_query

    q = apply_query(Application.query, Application, request)
    apps = q.query.all()
    return jsonify({'apps': [a.to_dict(fields=q.fields) for a in apps]})

Every parameter is optional: with none of them present this is a no-op and the
endpoint behaves exactly as it did before.
"""

import re
from dataclasses import dataclass
from datetime import datetime, timedelta

from sqlalchemy import and_, or_, inspect

from app.exceptions import ValidationError

# Hard ceiling on ``$top``. A list endpoint that answers "give me everything"
# is how /apps got to an unbounded ``query.all()``; a caller that genuinely
# wants more than this should page.
MAX_TOP = 500

COMPARE_OPS = {
    'eq': lambda col, val: col == val,
    'ne': lambda col, val: col != val,
    'gt': lambda col, val: col > val,
    'ge': lambda col, val: col >= val,
    'lt': lambda col, val: col < val,
    'le': lambda col, val: col <= val,
}


class QueryParseError(ValidationError):
    """A malformed query parameter.

    A ``ValidationError`` subclass so the global error handler answers it as a
    400 with the standard body — routes no longer need their own translation
    (the existing per-route ``except QueryParseError`` blocks keep working).
    """


@dataclass(frozen=True)
class ListQuery:
    """Typed representation of the shared list-query parameters.

    Existing endpoints may keep passing a Flask request to ``apply_query``.
    Schema-backed endpoints can parse this value once and pass it directly, so
    the query layer no longer requires a Flask request global.
    """

    select: str | None = None
    filter: str | None = None
    orderby: str | None = None
    skip: int = 0
    top: int | None = None

    def __post_init__(self):
        if self.skip < 0:
            raise QueryParseError('$skip cannot be negative.')
        if self.top is not None and self.top < 0:
            raise QueryParseError('$top cannot be negative.')
        if self.top is not None and self.top > MAX_TOP:
            object.__setattr__(self, 'top', MAX_TOP)

    @classmethod
    def from_args(cls, args, *, default_top=None):
        """Parse a MultiDict-like mapping into one validated value."""
        top = _int_arg(args.get('$top'), default_top, '$top')
        if top is not None:
            top = max(0, min(top, MAX_TOP))
        return cls(
            select=args.get('$select'),
            filter=args.get('$filter'),
            orderby=args.get('$orderby'),
            skip=_int_arg(args.get('$skip'), 0, '$skip'),
            top=top,
        )

    @classmethod
    def from_request(cls, http_request, *, default_top=None):
        return cls.from_args(http_request.args, default_top=default_top)


@dataclass(frozen=True)
class PageQuery:
    """Typed ``page``/``per_page`` parameters for offset-paginated lists.

    The page-numbered dialect of :class:`ListQuery` — some list endpoints and
    their consumers speak ``page``/``per_page`` rather than ``$skip``/``$top``.
    They are not interchangeable and are not pretended to be; this is the one
    place that dialect is parsed, defaulted, and bounded.
    """

    page: int = 1
    per_page: int = 50

    @property
    def skip(self) -> int:
        return (self.page - 1) * self.per_page

    @classmethod
    def from_args(cls, args, *, default_per_page=50, max_per_page=100):
        page = _int_arg(args.get('page'), 1, 'page')
        per_page = _int_arg(args.get('per_page'), default_per_page, 'per_page')
        return cls(page=max(1, page),
                   per_page=max(1, min(per_page, max_per_page)))

    @classmethod
    def from_request(cls, http_request, *, default_per_page=50, max_per_page=100):
        return cls.from_args(http_request.args, default_per_page=default_per_page,
                             max_per_page=max_per_page)


def _columns(model):
    """The model's real mapped columns — and the field allowlist."""
    return {c.key: getattr(model, c.key) for c in inspect(model).mapper.column_attrs}


def _column(model, field):
    field = (field or '').strip()
    cols = _columns(model)
    if field not in cols:
        raise QueryParseError(f"Unknown field '{field}' for '{model.__tablename__}'.")
    return cols[field]


def _coerce(value):
    """Literal -> Python. Quoted stays a string; bare words get typed."""
    value = value.strip()
    if len(value) >= 2 and value[0] == value[-1] and value[0] in ('"', "'"):
        return value[1:-1]
    low = value.lower()
    if low == 'true':
        return True
    if low == 'false':
        return False
    if low in ('null', 'none'):
        return None
    try:
        return int(value)
    except ValueError:
        pass
    try:
        return float(value)
    except ValueError:
        return value


# ---------------------------------------------------------------- $filter

_FUNCS = (
    ('contains', r"contains\(([^,]+),\s*['\"]([^'\"]*)['\"]\)",
     lambda col, v: col.ilike(f'%{v}%')),
    ('startswith', r"startswith\(([^,]+),\s*['\"]([^'\"]*)['\"]\)",
     lambda col, v: col.ilike(f'{v}%')),
    ('endswith', r"endswith\(([^,]+),\s*['\"]([^'\"]*)['\"]\)",
     lambda col, v: col.ilike(f'%{v}')),
)


def _parse_function(condition, model):
    """A function-call condition, or None if this isn't one."""
    for _name, pattern, build in _FUNCS:
        match = re.findall(pattern, condition)
        if match:
            field, value = match[0]
            return build(_column(model, field), value)

    match = re.findall(r'notempty\(([^)]+)\)', condition)
    if match:
        col = _column(model, match[0])
        return and_(col.isnot(None), col != '')

    match = re.findall(r"date\(([^,]+),\s*['\"]([^'\"]*)['\"]\)", condition)
    if match:
        field, value = match[0]
        try:
            day = datetime.strptime(value.strip(), '%Y-%m-%d')
        except ValueError as exc:
            raise QueryParseError(f"date() needs YYYY-MM-DD, got '{value}'.") from exc
        col = _column(model, field)
        return and_(col >= day, col < day + timedelta(days=1))

    match = re.findall(r'daysAgo\(([^,]+),\s*([^)]+)\)', condition)
    if match:
        field, raw = match[0]
        try:
            days = int(raw.strip())
        except ValueError as exc:
            raise QueryParseError(f"daysAgo() needs an integer, got '{raw}'.") from exc
        return _column(model, field) >= datetime.utcnow() - timedelta(days=days)

    return None


def _parse_condition(condition, model):
    built = _parse_function(condition, model)
    if built is not None:
        return built

    match = re.match(r'^\s*(\S+)\s+(eq|ne|gt|ge|lt|le)\s+(.+?)\s*$', condition)
    if not match:
        raise QueryParseError(f"Could not parse condition '{condition.strip()}'.")
    field, op, raw = match.groups()
    return COMPARE_OPS[op](_column(model, field), _coerce(raw))


def parse_filter(filter_str, model):
    """``$filter`` -> a WHERE clause, or None.

    One combiner per string (``and`` OR ``or``, not both) — same restriction as
    the original. Mixing them without parentheses is ambiguous, and silently
    picking a precedence is worse than refusing.
    """
    if not filter_str or not filter_str.strip():
        return None

    has_and = re.search(r'\s+and\s+', filter_str, re.IGNORECASE)
    has_or = re.search(r'\s+or\s+', filter_str, re.IGNORECASE)
    if has_and and has_or:
        raise QueryParseError("Mixing 'and' with 'or' needs separate requests — it is ambiguous here.")

    combiner, pattern = (or_, r'\s+or\s+') if has_or else (and_, r'\s+and\s+')
    parts = [p for p in re.split(pattern, filter_str, flags=re.IGNORECASE) if p.strip()]
    clauses = [_parse_condition(p, model) for p in parts]
    if not clauses:
        return None
    return clauses[0] if len(clauses) == 1 else combiner(*clauses)


# --------------------------------------------------------------- $orderby

def parse_orderby(orderby_str, model):
    """``$orderby=name asc,created_at desc`` -> a list of ORDER BY clauses."""
    if not orderby_str or not orderby_str.strip():
        return []
    out = []
    for part in orderby_str.split(','):
        bits = part.strip().split()
        if not bits:
            continue
        col = _column(model, bits[0])
        direction = (bits[1] if len(bits) > 1 else 'asc').lower()
        if direction not in ('asc', 'desc'):
            raise QueryParseError(f"Sort direction must be asc or desc, got '{direction}'.")
        out.append(col.desc() if direction == 'desc' else col.asc())
    return out


# ---------------------------------------------------------------- $select

def parse_select(select_str, model, extra=()):
    """``$select`` -> the requested field set, or None for "everything".

    `extra` names DERIVED fields a model's ``to_dict`` can emit that are not
    mapped columns (``project_name``, ``image_scan``, ``domains``…). They are
    accepted here so a caller can ask for one, and it is precisely by NOT
    asking that the caller skips its lazy load.
    """
    if not select_str or not select_str.strip():
        return None
    allowed = set(_columns(model)) | set(extra)
    fields = set()
    for raw in select_str.split(','):
        name = raw.strip()
        if not name:
            continue
        if name not in allowed:
            raise QueryParseError(f"Unknown field '{name}' for '{model.__tablename__}'.")
        fields.add(name)
    # The primary key always comes back — a row the client cannot identify is
    # not much use, and every UI keys its rows on it.
    pk = inspect(model).mapper.primary_key
    for col in pk:
        fields.add(col.key)
    return fields


# ------------------------------------------------------------- apply_query

class QueryResult:
    """The narrowed query plus what the caller needs to serialize it."""

    __slots__ = ('query', 'fields', 'skip', 'top', 'total')

    def __init__(self, query, fields, skip, top, total):
        self.query = query
        self.fields = fields
        self.skip = skip
        self.top = top
        self.total = total

    def as_meta(self):
        """Paging metadata for the response envelope."""
        return {'total': self.total, 'skip': self.skip, 'top': self.top}


def apply_query(query, model, request_or_query, *, default_top=None, select_extra=()):
    """Apply the query params present on `request` to `query`.

    ``request_or_query`` accepts either Flask's request object or a pre-parsed
    :class:`ListQuery` from a schema-backed endpoint.

    Returns a :class:`QueryResult`. With no params present this is a no-op and
    ``.fields`` is None, so an endpoint can adopt it without changing any
    existing caller's response.

    `default_top` bounds an endpoint that has no explicit ``$top``; leave it
    None to keep today's unbounded behaviour while callers migrate.
    """
    list_query = (
        request_or_query
        if isinstance(request_or_query, ListQuery)
        else ListQuery.from_request(request_or_query, default_top=default_top)
    )

    where = parse_filter(list_query.filter, model)
    if where is not None:
        query = query.filter(where)

    order = parse_orderby(list_query.orderby, model)
    if order:
        query = query.order_by(*order)

    fields = parse_select(list_query.select, model, extra=select_extra)

    # `total` is the count BEFORE paging — the client needs it to render "12 of
    # 340", and computing it after .limit() would just return the page size.
    total = None
    skip = list_query.skip
    top = list_query.top
    if skip or top is not None:
        total = query.order_by(None).count()
        if skip:
            query = query.offset(skip)
        if top is not None:
            query = query.limit(top)

    return QueryResult(query, fields, skip, top, total)


def _int_arg(raw, fallback, name):
    if raw is None or raw == '':
        return fallback
    try:
        value = int(raw)
    except (TypeError, ValueError) as exc:
        raise QueryParseError(f'{name} must be an integer.') from exc
    if value < 0:
        raise QueryParseError(f'{name} cannot be negative.')
    return value