"""Resolves a Dataset's rows: a Master's records or a Form's submissions, with the query's filters, sort, column selection or group-by aggregation applied. The filters, sort, paging and aggregates run in SQL wherever app/crud/record_sql.py can express them exactly (nearly every query), so a query reads only the rows or groups it returns. Anything it can't express exactly — a custom metric expression, a sort over linked-record labels — still has its WHERE narrowed in SQL and is finished by the Python evaluation in this module, which remains the reference semantics record_sql reproduces.""" import re import uuid from dataclasses import dataclass from datetime import date, datetime, timedelta from typing import Any from sqlalchemy import case, func, literal_column, select from sqlalchemy.ext.asyncio import AsyncSession from app.crud import record_sql, summary_cache from app.crud.filter_tree import conditions, date_kind_for, map_conditions, passes, resolve_relative_dates from app.crud.record_sql import coerce_date as _coerce_date from app.crud.scope import _record_scope_values, is_restricted_for from app.models.master import Master, MasterRecord from app.models.workflow import Form, FormSubmission # Form field types whose value is an array of linked-record ids rather than a # single one — mirrors frontend/src/lib/types.ts's FORM_MULTI_VALUE_FIELD_TYPES. _FORM_MULTI_VALUE_TYPES = {"multiselect", "treeselect"} # System-managed audit columns every Master record / Form submission carries # (see db/schema.sql) — merged into each row dict under these uniform names # regardless of source_type, but deliberately excluded from the "no explicit # columns requested" column_names union below so an already-published page # that never asked for them keeps rendering exactly as it did before this # existed. A page author opts in by adding one of these names to the block's # own `columns` list (see PageDesigner.tsx's availableColumns). _AUDIT_COLUMN_NAMES = {"created_at", "updated_at", "created_by", "updated_by"} def _audit_columns(created_at: Any, updated_at: Any, created_by: Any, updated_by: Any) -> dict: return { "created_at": created_at.isoformat() if created_at else None, "updated_at": updated_at.isoformat() if updated_at else None, "created_by": str(created_by) if created_by else None, "updated_by": str(updated_by) if updated_by else None, } # ---- "Current user" / session tokens in filter values --------------------------------------------- # # A saved dataset is shared (dashboards, pages, other users), so a filter # like "created_by equals the viewer" must be resolved per request, here, not # by whoever built it. A filter value that is exactly one of these tokens is # swapped for the viewing user's value before filtering: # {{current_user.id}} / .name / .email / .vendor_id / .tenant_id # {{session.}} — the viewer's sign-in session value (Users → Session data) # The braces are optional, and "current.userid", "currentUserId", # "current_user_id" etc. all read as current_user.id; "{{me}}" is shorthand. _USER_TOKENS = { "currentuserid": "id", "currentuser": "id", "me": "id", "userid": "id", "currentusername": "name", "currentuseremail": "email", "currentuservendorid": "vendor_id", "currentusertenantid": "tenant_id", } _NO_USER = "\x00no-signed-in-user" # matches nothing def _token_of(value: Any) -> tuple[str, str] | None: """("user", field) / ("session", key) when `value` is a whole token.""" if not isinstance(value, str): return None text = value.strip() braced = text.startswith("{{") and text.endswith("}}") if braced: text = text[2:-2].strip() if text.lower().startswith("session.") and len(text) > 8: return ("session", text[8:]) if braced or "session." in value else None key = re.sub(r"[^a-z0-9]", "", text.lower()) field = _USER_TOKENS.get(key) # Unbraced, only the unmistakable current-user spellings count — a bare # "me" or "userid" might be real data. if field and (braced or ("current" in key and "user" in key)): return ("user", field) return None async def resolve_filter_tokens(db: AsyncSession, tenant_id: uuid.UUID, user_id: uuid.UUID | None, filters: list[dict]) -> list[dict]: def values(v: Any) -> list: return v if isinstance(v, list) else [v] if not any(_token_of(v) for f in conditions(filters) for v in values(f.get("value"))): return filters from app.models.identity import Tenant, User user = await db.get(User, user_id) if user_id else None if user is not None and user.tenant_id != tenant_id: user = None session: dict | None = None async def resolve(v: Any) -> Any: nonlocal session token = _token_of(v) if token is None: return v if user is None: return _NO_USER kind, key = token if kind == "user": value = getattr(user, key, None) return str(value) if value is not None else _NO_USER if session is None: from app.crud.login_session import build_login_session_data from app.schemas.identity import LoginSessionSettings # The configured lookups only — never the sign-in task, which may # have side effects and isn't meant to run on every query. tenant = await db.get(Tenant, tenant_id) try: settings = LoginSessionSettings.model_validate((tenant.login_session_settings if tenant else None) or {}).model_copy(update={"task_id": None}) session = await build_login_session_data(db, tenant, user, settings) if tenant is not None else {} except ValueError: session = {} value = session.get(key) return _NO_USER if value is None else value resolved_values: dict[int, Any] = {} for f in conditions(filters): raw = f.get("value") resolved_values[id(f)] = [await resolve(v) for v in raw] if isinstance(raw, list) else await resolve(raw) return map_conditions(filters, lambda f: {**f, "value": resolved_values[id(f)]}) async def prepare_filters( db: AsyncSession, tenant_id: uuid.UUID, user_id: uuid.UUID | None, filters: list[dict], field_defs: list[dict] | None = None ) -> list[dict]: """A filter tree ready to run: the viewer's {{tokens}} resolved and each relative date ("today", "this month"…) turned into its range in the tenant's time zone. `field_defs` (the source's fields) say which fields hold plain dates.""" filters = await resolve_filter_tokens(db, tenant_id, user_id, filters) # "is one of" typed into a filter box: "Open, Closed" is two values. filters = map_conditions( filters, lambda f: {**f, "value": [v.strip() for v in f["value"].split(",") if v.strip()]} if f.get("operator") in ("in", "not_in") and isinstance(f.get("value"), str) else f, ) if any(f.get("operator") == "date_is" for f in conditions(filters)): from app.core.timezones import tenant_zone filters = resolve_relative_dates(filters, await tenant_zone(db, tenant_id), date_kind_for(field_defs or [])) return filters # ---- Created by / Updated by: stored as user ids, shown and matched by name ---------------------- _AUDIT_USER_FIELDS = ("created_by", "updated_by") async def _user_labels(db: AsyncSession, tenant_id: uuid.UUID, ids: set | None = None) -> dict[str, tuple[str, str]]: """{user id: (name, email)} for the tenant's users — or just `ids`, when only some rows' people need naming.""" from app.models.identity import User stmt = select(User.id, User.name, User.email).where(User.tenant_id == tenant_id) if ids is not None: wanted = [] for raw in ids: try: wanted.append(raw if isinstance(raw, uuid.UUID) else uuid.UUID(str(raw))) except ValueError: continue if not wanted: return {} stmt = stmt.where(record_sql.ids_any(User.id, wanted)) rows = (await db.execute(stmt)).all() return {str(uid): (name or "", email or "") for uid, name, email in rows} def _user_filter_passes(operator: str, raw: Any, expected: Any, labels: dict[str, tuple[str, str]]) -> bool: """A Created by / Updated by filter matches the user's id, name or email — so "created_by equals {{current_user.id}}" and "created_by contains Priya" both work.""" if operator in ("is_empty", "is_not_empty", "is_true", "is_false"): return _eval_operator(operator, raw, expected) candidates = [raw, *labels.get(str(raw), ())] if raw is not None else [] # "does not equal / is not one of / does not contain": none of id, name or email does. negated = {**NEGATED_OPERATORS, "not_equals": "equals"} if operator in negated: return not any(_eval_operator(negated[operator], c, expected) for c in candidates) return any(_eval_operator(operator, c, expected) for c in candidates) def _condition_passes(f: dict, row: dict, audit_user_fields: set[str], labels: dict[str, tuple[str, str]] | None) -> bool: if f["field"] in audit_user_fields: return _user_filter_passes(f["operator"], row.get(f["field"]), f.get("value"), labels or {}) return _eval_operator(f["operator"], row.get(f["field"]), f.get("value")) def _row_passes(f: dict, row: dict, audit_user_fields: set[str], labels: dict[str, tuple[str, str]] | None) -> bool: """One filter-list item — a condition, or a group of them (any/all).""" return passes([f], lambda c: _condition_passes(c, row, audit_user_fields, labels)) async def filter_records( db: AsyncSession, tenant_id: uuid.UUID, user_id: uuid.UUID | None, records: list, filters: list[dict], declared_names: set[str], field_defs: list[dict] | None = None, ) -> list: """The dataset filter semantics (operators, audit columns, per-viewer {{current_user.id}} tokens) applied to MasterRecord / FormSubmission objects — for the plain list endpoints a Page's live-paginated Data List uses (routers/masters.py, routers/forms.py), so its block filters behave exactly as they do everywhere else.""" if not filters: return records filters = await prepare_filters(db, tenant_id, user_id, filters, field_defs) audit_user_fields = {f for f in _AUDIT_USER_FIELDS if f not in declared_names} labels = await _user_labels(db, tenant_id) if any(f["field"] in audit_user_fields for f in conditions(filters)) else None kept = [] for rec in records: created_by = getattr(rec, "created_by", None) or getattr(rec, "submitted_by", None) row = {**(rec.field_values or {}), "id": str(rec.id)} for k, v in _audit_columns(rec.created_at, rec.updated_at, created_by, rec.updated_by).items(): if k not in declared_names: row[k] = v if all(_row_passes(f, row, audit_user_fields, labels) for f in filters): kept.append(rec) return kept async def attach_user_names(db: AsyncSession, tenant_id: uuid.UUID, records: list) -> None: """Sets created_by_name / updated_by_name on each record (read by MasterRecordRead / FormSubmissionRead) so a list can show people, not ids.""" if not records: return people = {getattr(rec, "created_by", None) or getattr(rec, "submitted_by", None) for rec in records} labels = await _user_labels(db, tenant_id, (people | {rec.updated_by for rec in records}) - {None}) def name(uid: Any) -> str | None: label = labels.get(str(uid)) if uid else None return (label[0] or label[1]) if label else None for rec in records: rec.created_by_name = name(getattr(rec, "created_by", None) or getattr(rec, "submitted_by", None)) rec.updated_by_name = name(rec.updated_by) def parse_filters_param(raw: str | None) -> list[dict]: """The `filters` query parameter: a JSON list of {field, operator, value} — the same shape a dataset stores. ValueError when malformed.""" if not raw: return [] import json from pydantic import TypeAdapter from app.schemas.reporting import DatasetFilterItem try: parsed = TypeAdapter(list[DatasetFilterItem]).validate_python(json.loads(raw)) except Exception as exc: # noqa: BLE001 - JSON or validation error, reported as one message raise ValueError(f"filters must be a JSON list of {{field, operator, value}}: {exc}") from exc return [f.model_dump() for f in parsed] # "not in" / "does not contain": a value that's there and isn't / doesn't — # like "does not equal", a missing value never matches. NEGATED_OPERATORS = {"not_in": "in", "not_contains": "contains"} _TRUE_TEXT = {"true", "yes", "y", "1", "on"} def is_truthy(value: Any) -> bool: """A yes/no reading of a value: True, a non-zero number, or text such as "true"/"yes"/"1". Anything else — missing included — is "no".""" if isinstance(value, bool): return value if isinstance(value, (int, float)): return value != 0 if isinstance(value, str): return value.lower() in _TRUE_TEXT return False def _affix(operator: str, actual: Any, expected: Any) -> bool: """starts_with / ends_with, case-insensitive; a list matches when any of its items does.""" pattern = str(expected).lower() if isinstance(actual, dict): return False items = actual if isinstance(actual, list) else [actual] test = str.startswith if operator == "starts_with" else str.endswith return any(test(str(v).lower(), pattern) for v in items) def _eval_operator(operator: str, actual: Any, expected: Any) -> bool: if operator == "is_empty": return actual is None or actual == "" if operator == "is_not_empty": return actual is not None and actual != "" if operator == "is_true": return is_truthy(actual) if operator == "is_false": return not is_truthy(actual) if actual is None: return False if operator in NEGATED_OPERATORS: return not _eval_operator(NEGATED_OPERATORS[operator], actual, expected) if operator in ("starts_with", "ends_with"): return _affix(operator, actual, expected) try: if operator == "equals": return actual == expected if operator == "not_equals": return actual != expected if operator == "gt": return actual > expected if operator == "gte": return actual >= expected if operator == "lt": return actual < expected if operator == "lte": return actual <= expected if operator == "in": return actual in (expected if isinstance(expected, list) else [expected]) if operator == "contains": return str(expected).lower() in str(actual).lower() if operator == "between": lo, hi = expected actual_d, lo_d, hi_d = _coerce_date(actual), _coerce_date(lo), _coerce_date(hi) # A date-only upper bound means "through the end of that day". if isinstance(hi, str) and len(hi.strip()) == 10 and isinstance(hi_d, datetime): return lo_d <= actual_d < hi_d + timedelta(days=1) return lo_d <= actual_d <= hi_d except TypeError: return False return False def _bucket_date(value: Any, bucket: str) -> str: """Buckets a raw field value into its day/week/month label for chart x-axis grouping (Phase 3's x_bucket) — a value that isn't a parseable date falls into "(invalid)" rather than raising, consistent with this module's "never raise on loosely-typed JSONB" convention elsewhere (_sort_key, _eval_operator's own except TypeError).""" coerced = _coerce_date(value) if not isinstance(coerced, (datetime, date)): return "(invalid)" d = coerced.date() if isinstance(coerced, datetime) else coerced if bucket == "day": return d.isoformat() if bucket == "week": return (d - timedelta(days=d.weekday())).isoformat() if bucket == "month": return f"{d.year:04d}-{d.month:02d}" return d.isoformat() def _sort_key(value: Any) -> tuple: # Buckets numeric-parseable values (compared numerically) ahead of # everything else (compared as strings), with None last — tuple # comparison never touches the second element across buckets, so mixed # types in a loosely-typed JSONB column can't raise TypeError. if value is None: return (2, "") try: return (0, float(value)) except (TypeError, ValueError): return (1, str(value)) async def _apply_reference_labels( db: AsyncSession, tenant_id: uuid.UUID, rows: list[dict], ref_fields: dict[str, tuple], ) -> None: """Mutates `rows` in place: for each field in `ref_fields` (name -> (linked_id, display_field, multi[, is_form])), the raw linked-record id(s) are replaced with that record's display value. `is_form` marks a link to a Form's submissions rather than a Master's records. Shared by both callers below — a Master's own "reference" fields and a Form's Master-linked choice fields resolve the same way once each has produced its own (field -> linked master + display field) map.""" for field_name, spec in ref_fields.items(): ref_master_id, display_field, multi = spec[:3] is_form = len(spec) > 3 and bool(spec[3]) if not ref_master_id or not display_field: continue # Only the linked records these rows point at — not the whole table. linked_ids: set[uuid.UUID] = set() for row in rows: raw = row.get(field_name) for v in raw if isinstance(raw, list) else [raw]: if v is None: continue try: linked_ids.add(uuid.UUID(str(v))) except ValueError: continue if not linked_ids: continue if is_form: stmt = select(FormSubmission.id, FormSubmission.field_values).where( FormSubmission.tenant_id == tenant_id, FormSubmission.form_id == uuid.UUID(str(ref_master_id)), FormSubmission.is_deleted.is_(False), record_sql.ids_any(FormSubmission.id, sorted(linked_ids)), ) else: stmt = select(MasterRecord.id, MasterRecord.field_values).where( MasterRecord.tenant_id == tenant_id, MasterRecord.master_id == uuid.UUID(str(ref_master_id)), MasterRecord.is_deleted.is_(False), record_sql.ids_any(MasterRecord.id, sorted(linked_ids)), ) linked = (await db.execute(stmt)).all() label_by_id = {str(rid): (fv.get(display_field) or str(rid)) for rid, fv in linked} for row in rows: raw = row.get(field_name) if raw is None: continue if multi and isinstance(raw, list): row[field_name] = ", ".join(label_by_id.get(str(v), str(v)) for v in raw) elif not multi: row[field_name] = label_by_id.get(str(raw), str(raw)) async def _resolve_reference_fields( db: AsyncSession, tenant_id: uuid.UUID, master: Master, rows: list[dict], field_names: set[str] | None ) -> None: """For any field in `field_names` (or every reference field on the table when `field_names` is None — "no column restriction," the same convention an empty `columns` list uses elsewhere) that is a "reference" field on `master`'s schema (a dropdown live-linked to another Master's records — see MasterFieldDef.reference_master_id), resolve it via _apply_reference_labels. This is what makes "count packages by country" or "states with their country name" work without a real join engine: a reference field IS the tenant's own declared link between two tables, so resolving it in place reads as pulling data from the linked table without needing arbitrary multi-table joins.""" ref_fields = { f["name"]: ( f.get("reference_form_id") or f.get("reference_master_id"), f.get("reference_display_field"), bool(f.get("multi")), bool(f.get("reference_form_id")), ) for f in master.field_schema if f.get("type") == "reference" and (field_names is None or f["name"] in field_names) } if ref_fields: await _apply_reference_labels(db, tenant_id, rows, ref_fields) async def _resolve_form_reference_fields( db: AsyncSession, tenant_id: uuid.UUID, form: Form, rows: list[dict], field_names: set[str] | None ) -> None: """Same idea as _resolve_reference_fields, but for a Form's own Master-linked choice fields (FR-FRM-004: dropdown/autocomplete/lookup/ etc. backed by a live Master instead of static options — see FormFieldDef.master_id/master_display_field). A submission stores the linked record's raw id (FormFieldRenderer.tsx's useChoiceOptions uses the record's id as the option value), so without this a form-sourced dataset/detail view would show a bare UUID instead of e.g. the linked employee's name.""" ref_fields = { f["name"]: (str(f["master_id"]), f.get("master_display_field"), f.get("type") in _FORM_MULTI_VALUE_TYPES) for f in form.field_definitions if f.get("master_id") and (field_names is None or f["name"] in field_names) } # (A link_source field keeps its raw id here — row actions pass it on, # e.g. as recordId — and pages show its record's name themselves.) if ref_fields: await _apply_reference_labels(db, tenant_id, rows, ref_fields) def _aggregate_value(fn: str, values: list[Any]) -> float: nums: list[float] = [] for v in values: try: nums.append(float(v)) except (TypeError, ValueError): continue if fn == "sum": return sum(nums) if fn == "avg": return round(sum(nums) / len(nums), 4) if nums else 0.0 if fn == "min": return min(nums) if nums else 0.0 if fn == "max": return max(nums) if nums else 0.0 return 0.0 def _aggregate_rows( rows: list[dict], group_by: str, aggregate_field: str | None, aggregate_fn: str, sort_field: str | None, sort_dir: str, ) -> tuple[list[dict], list[str]]: output_key = "count" if aggregate_fn == "count" else f"{aggregate_fn}_{aggregate_field}" groups: dict[str, list[dict]] = {} order: list[str] = [] for r in rows: raw = r.get(group_by) label = "(empty)" if raw is None or raw == "" else str(raw) if label not in groups: groups[label] = [] order.append(label) groups[label].append(r) result = [ { group_by: label, output_key: len(groups[label]) if aggregate_fn == "count" else _aggregate_value(aggregate_fn, [r.get(aggregate_field) for r in groups[label]]), } for label in order ] _sort_aggregated(result, group_by, output_key, sort_field, sort_dir) return result, [group_by, output_key] def _sort_aggregated(result: list[dict], group_by: str, output_key: str, sort_field: str | None, sort_dir: str) -> None: if sort_field == group_by: result.sort(key=lambda r: _sort_key(r[group_by]), reverse=(sort_dir == "desc")) else: # Unset, or explicitly the aggregate column: rank by the computed # value — descending by default, since "biggest group first" is # what a count/sum query is almost always asked for. result.sort(key=lambda r: r[output_key], reverse=(sort_dir != "asc")) # ---- Query execution ------------------------------------------------------------------------------ # # Every query builds its WHERE in SQL first (app/crud/record_sql.py). When # that WHERE is exact, the rest — count, sort, page, group-by aggregates — # runs in SQL too, and only the rows or groups actually returned are read. # When any part isn't exact (or the query needs every row, like a custom # expression), the WHERE still narrows what's fetched and the Python # evaluation finishes the job exactly as before. @dataclass class _Query: src: record_sql.RecordSource filters: list[dict] # {{tokens}} already resolved for the viewer where: list exact: bool allowed_scope: set[str] | None # Form fields hidden to the viewer (app/crud/field_access.py) — left out # of every row; filtering, sorting or grouping by one is refused. hidden: frozenset[str] = frozenset() # Lookup columns ("customer.city") this query reads — see _attach_lookups. # Any at all means the rows are finished in Python. lookups: frozenset[str] = frozenset() # Calculated columns ({name, expression}) worked out on each row, after # its lookups — see _compute_columns. Also finished in Python. computed: tuple = () # Rollup columns ("rollup:count:form::application_id") — values from # the records that link to each row; see _attach_rollups. rollups: frozenset[str] = frozenset() user_id: uuid.UUID | None = None # Other sources joined onto each row ("office.city") — see _apply_joins. joins: tuple = () def refuse_hidden(self, *fields: str | None) -> None: blocked = sorted({f for f in fields if f} & self.hidden) if blocked: from fastapi import HTTPException raise HTTPException(status_code=403, detail=f"You don't have access to {', '.join(blocked)}.") def strip(self, rows: list[dict]) -> list[dict]: if not self.hidden: return rows return [{k: v for k, v in r.items() if k not in self.hidden and k.split(".")[0] not in self.hidden} for r in rows] def use_lookups(self, names: set[str]) -> None: if names: self.refuse_hidden(*(n.split(".")[0] for n in names)) self.lookups = self.lookups | frozenset(names) self.exact = False def use_rollups(self, names: set[str]) -> None: if names: self.rollups = self.rollups | frozenset(names) self.exact = False def use_computed(self, computed: list[dict], src: record_sql.RecordSource) -> None: """Calculated columns: the fields they read must be readable (a hidden one can't leak through a formula), and the lookups they read are read.""" from app.core.expr import expression_references names = {c["name"] for c in computed} refs = {r for c in computed for r in expression_references(c["expression"]) if r not in names} self.refuse_hidden(*(r.split(".")[0] for r in refs if not is_rollup(r))) self.use_lookups({r for r in refs if _is_lookup(src, r)}) self.use_rollups({r for r in refs if is_rollup(r)}) self.computed = tuple(computed) self.exact = False # Field types whose values are numbers — a filter value typed for one ("100" # from a filter box) is compared as the number it spells. _NUMERIC_TYPES = {"number", "integer", "decimal", "currency", "percentage", "slider", "rating", "bigint", "smallint", "numeric", "real", "double_precision"} _NUMBER_TEXT = re.compile(r"^\s*-?\d+(\.\d+)?\s*$") def _as_number(value: Any) -> Any: if isinstance(value, str) and _NUMBER_TEXT.match(value): n = float(value) return int(n) if n.is_integer() and "." not in value else n if isinstance(value, list): return [_as_number(v) for v in value] return value def coerce_filter_numbers(filters: list[dict], numeric_fields: set[str]) -> list[dict]: """Number-like text filter values on numeric fields, as numbers.""" if not numeric_fields: return filters return map_conditions(filters, lambda f: {**f, "value": _as_number(f.get("value"))} if f.get("field") in numeric_fields else f) async def vendor_of(db: AsyncSession, user_id: uuid.UUID | None) -> uuid.UUID | None: """The Portal vendor a user belongs to, or None for internal users.""" if user_id is None: return None from app.models.identity import User return (await db.execute(select(User.vendor_id).where(User.id == user_id))).scalar_one_or_none() async def _prepare_query( db: AsyncSession, tenant_id: uuid.UUID, source_type: str, source_id: uuid.UUID, filters: list[dict], user_id: uuid.UUID | None, computed: list[dict] | None = None, joins: list[dict] | None = None, ) -> _Query: # A table datasource is read through its typed table (record_sql.with_tables). src = await record_sql.load_source(db, tenant_id, source_type, source_id, typed=True) aliases = _check_join_aliases(src, joins) filters = await prepare_filters(db, tenant_id, user_id, filters, src.field_defs) filters = coerce_filter_numbers(filters, {f["name"] for f in src.field_defs if f.get("name") and f.get("type") in _NUMERIC_TYPES}) where = src.base_where(tenant_id) # A Portal (vendor) user reads only their own organisation's submissions — # the same rule the submission endpoints apply. (The database's # vendor_isolation policy doesn't bind the table-owning app role.) if src.kind == "form": vendor = await vendor_of(db, user_id) if vendor is not None: where.append((src.table.c.vendor_id if src.table is not None else FormSubmission.vendor_id) == vendor) allowed: set[str] | None = None if src.master is not None and src.master.scope_field: allowed = await is_restricted_for(db, tenant_id, user_id, src.master) if allowed is not None: where.append(record_sql.scope_clause(src.fv, src.master.scope_field, allowed)) # Record-level access policy (app/crud/access_policy.py). from app.crud.access_policy import access_clause restriction = await access_clause(db, tenant_id, user_id, src, "view") if restriction is not None: where.append(restriction) hidden: frozenset[str] = frozenset() if src.kind == "form": from app.crud import field_access access = await field_access.access_for(db, tenant_id, user_id, src.field_defs, with_step_rules=False) hidden = frozenset(field_access.hidden_names(access)) # Lookup and calculated columns are worked out after the rows are read, # so a condition on one is checked in Python. virtual = ( {c["name"] for c in computed or []} | {f["field"] for f in conditions(filters) if is_rollup(f["field"])} | {f["field"] for f in conditions(filters) if f["field"].split(".")[0] in aliases} ) lookup_filters = {f["field"] for f in conditions(filters) if _is_lookup(src, f["field"])} clauses, exact = record_sql.filter_clauses(src, filters, tenant_id, skip=lookup_filters | virtual) query = _Query(src, filters, where + clauses, exact, allowed, hidden, user_id=user_id) query.refuse_hidden(*(f["field"].split(".")[0] for f in conditions(filters) if f["field"] not in virtual)) query.use_lookups(lookup_filters) query.use_rollups({f for f in virtual if is_rollup(f)}) if joins: query.joins = tuple(joins) query.exact = False if computed: query.use_computed(computed, src) return query # ---- Lookup columns ------------------------------------------------------------------------------- # A column named "." reads from the record a link # field points at — a Master's "reference" field or a Form's datasource-linked # choice field — e.g. "customer.city" on Orders whose "customer" links to # Customers. Many-to-one, resolved for just the rows being returned; a # multi-valued link lists its records' values comma-separated. def _links(src: record_sql.RecordSource) -> dict[str, tuple[str, bool, bool]]: """link field name -> (linked master/form id, links to a form, multi-valued).""" out: dict[str, tuple[str, bool, bool]] = {} for f in src.field_defs: if src.kind == "master" and f.get("type") == "reference" and (f.get("reference_form_id") or f.get("reference_master_id")): out[f["name"]] = (str(f.get("reference_form_id") or f.get("reference_master_id")), bool(f.get("reference_form_id")), bool(f.get("multi"))) elif src.kind == "form" and f.get("master_id"): out[f["name"]] = (str(f["master_id"]), False, f.get("type") in _FORM_MULTI_VALUE_TYPES) elif src.kind == "form" and f.get("link_source"): kind, _, linked = str(f["link_source"]).partition(":") out[f["name"]] = (linked, kind == "form", False) return out def _is_lookup(src: record_sql.RecordSource, name: str | None) -> bool: if not name or "." not in name or name in src.declared: return False link, _, target = name.partition(".") return bool(target) and link in _links(src) # How many links a lookup column may follow: "application_id.office_id.district" # is two. MAX_LOOKUP_HOPS = 3 async def _attach_lookups( db: AsyncSession, tenant_id: uuid.UUID, src: record_sql.RecordSource, rows: list[dict], names: frozenset[str], hops: int = 1 ) -> None: """Adds each lookup column to `rows`, reading the raw link ids — so this runs before _decorate_rows turns those ids into labels. A column that follows more links ("application_id.office_id.district") is read the same way from the linked records, one hop at a time.""" links = _links(src) targets: dict[str, set[str]] = {} for name in names: link, _, target = name.partition(".") if link in links and target: targets.setdefault(link, set()).add(target) for link, wanted in targets.items(): linked_id, is_form, multi = links[link] ids: set[uuid.UUID] = set() for row in rows: raw = row.get(link) for v in raw if isinstance(raw, list) else [raw]: try: ids.add(uuid.UUID(str(v))) except ValueError: continue linked: dict[str, dict] = {} if ids: model = FormSubmission if is_form else MasterRecord parent = FormSubmission.form_id if is_form else MasterRecord.master_id creator = FormSubmission.submitted_by if is_form else MasterRecord.created_by stmt = select(model.id, model.field_values, model.created_at, model.updated_at, creator, model.updated_by).where( model.tenant_id == tenant_id, parent == uuid.UUID(linked_id), model.is_deleted.is_(False), record_sql.ids_any(model.id, sorted(ids)) ) # The linked record's own system columns too — "application.created_by" # (who submitted it), created_at… — under the same names as a row's # own (_row_dict); a field the record itself declares keeps its value. linked = { str(rid): {**_audit_columns(c_at, u_at, c_by, u_by), **(fv or {}), "id": str(rid)} for rid, fv, c_at, u_at, c_by, u_by in (await db.execute(stmt)).all() } # The next hop ("office_id.district" on the linked applications), # read onto the linked records first. deeper = frozenset(t for t in wanted if "." in t) if deeper and linked and hops < MAX_LOOKUP_HOPS: linked_src = await record_sql.load_source(db, tenant_id, "form" if is_form else "master", uuid.UUID(linked_id)) await _attach_lookups(db, tenant_id, linked_src, list(linked.values()), deeper, hops + 1) for row in rows: raw = row.get(link) records = [linked.get(str(v)) for v in (raw if isinstance(raw, list) else [raw]) if v is not None] for target in wanted: values = [r.get(target) for r in records if r is not None and r.get(target) is not None] row[f"{link}.{target}"] = ", ".join(str(v) for v in values) if multi else (values[0] if values else None) # A looked-up field that is itself a link (a seat's office) shows its # linked record's label, like the source's own link fields do. target_defs = await _linked_field_defs(db, tenant_id, linked_id, is_form) nested = {} for target in wanted: spec = _label_spec(target_defs.get(target)) if spec and not multi: nested[f"{link}.{target}"] = spec if nested: await _apply_reference_labels(db, tenant_id, rows, nested) def _label_spec(f: dict | None) -> tuple | None: """How a link field's stored id is shown — (linked source id, display field, multi[, is_form]) for _apply_reference_labels — or None.""" if not f: return None if f.get("type") == "reference" and (f.get("reference_form_id") or f.get("reference_master_id")): return (f.get("reference_form_id") or f.get("reference_master_id"), f.get("reference_display_field"), bool(f.get("multi")), bool(f.get("reference_form_id"))) if f.get("master_id"): return (str(f["master_id"]), f.get("master_display_field"), f.get("type") in _FORM_MULTI_VALUE_TYPES) if f.get("link_source") and f.get("link_display_field"): kind, _, linked = str(f["link_source"]).partition(":") return (linked, f["link_display_field"], False, kind == "form") return None # ---- Rollup columns ---------------------------------------------------------------------------- # The other direction: values from the records that link TO each row. A # column named # rollup::::[:] # e.g. "rollup:count:form::application_id" on Applications # is each application's number of replies. fn: count, sum, avg, min, max, # latest (the field's value on the newest child — its date when no field is # named) or list (the distinct values, comma-separated). Only the children # the viewer may see count — they're read through the same access rules. ROLLUP_FUNCTIONS = ("count", "sum", "avg", "min", "max", "latest", "list") # Children read per rollup at most — a guard, not a paging limit. MAX_ROLLUP_CHILDREN = 50_000 def parse_rollup(name: str | None) -> tuple[str, str, uuid.UUID, str, str | None] | None: """(fn, child kind, child id, link field, field) — or None.""" if not name or not name.startswith("rollup:"): return None parts = name.split(":") if len(parts) not in (5, 6): return None _, fn, kind, child_id, link = parts[:5] field = parts[5] if len(parts) == 6 and parts[5] else None if fn not in ROLLUP_FUNCTIONS or kind not in ("master", "form") or not link: return None try: child = uuid.UUID(child_id) except ValueError: return None if fn in ("sum", "avg", "min", "max", "list") and not field: return None return fn, kind, child, link, field def is_rollup(name: str | None) -> bool: return parse_rollup(name) is not None def _rollup_value(fn: str, field: str | None, children: list[dict]) -> Any: if fn == "count": return len(children) if fn == "latest": if not children: return None newest = max(children, key=lambda c: c.get("created_at") or "") return newest.get(field or "created_at") values = [c.get(field) for c in children if c.get(field) not in (None, "")] if fn == "list": return ", ".join(dict.fromkeys(str(v) for v in values)) or None if fn in ("sum", "avg"): return _aggregate_value(fn, values) if values else (0 if fn == "sum" else None) # min / max: numbers as numbers, anything else (dates) as text. if not values: return None try: nums = [float(v) for v in values] best = min(nums) if fn == "min" else max(nums) return int(best) if best.is_integer() else best except (TypeError, ValueError): texts = [str(v) for v in values] return min(texts) if fn == "min" else max(texts) async def _attach_rollups( db: AsyncSession, tenant_id: uuid.UUID, src: record_sql.RecordSource, rows: list[dict], names: frozenset[str], user_id: uuid.UUID | None ) -> None: """Adds each rollup column to `rows` (see above): one read of the child records per (child source, link field), grouped by the row they link to.""" groups: dict[tuple[str, uuid.UUID, str], list[tuple[str, str, str | None]]] = {} for name in names: parsed = parse_rollup(name) if parsed: fn, kind, child_id, link, field = parsed groups.setdefault((kind, child_id, link), []).append((name, fn, field)) ids = [r["id"] for r in rows if r.get("id")] for (kind, child_id, link), wanted in groups.items(): by_parent: dict[str, list[dict]] = {} if ids: child_src = await record_sql.load_source(db, tenant_id, kind, child_id) link_def = child_src.field_def(link) or {} multi = bool(link_def.get("multi")) or link_def.get("type") in _FORM_MULTI_VALUE_TYPES # A one-to-one link is narrowed to these rows' ids in SQL; a # multi-valued one (or very many rows) is matched here. filters = [] if multi or len(ids) > 500 else [{"field": link, "operator": "in", "value": ids}] child_query = await _prepare_query(db, tenant_id, kind, child_id, filters, user_id) child_query.refuse_hidden(link, *(f for _, _, f in wanted if f)) stmt = select(*_record_columns(child_query.src)).where(*child_query.where).order_by(*child_query.src.default_order()).limit(MAX_ROLLUP_CHILDREN) children = [_row_dict(child_query.src.declared, *r) for r in (await db.execute(record_sql.with_tables(child_query.src, stmt))).all()] if not child_query.exact: children = [c for c in children if passes(child_query.filters, lambda f, c=c: _condition_passes(f, c, set(), None))] wanted_ids = set(ids) for child in children: raw = child.get(link) for parent in raw if isinstance(raw, list) else [raw]: if parent is not None and str(parent) in wanted_ids: by_parent.setdefault(str(parent), []).append(child) for row in rows: children = by_parent.get(str(row.get("id")), []) for name, fn, field in wanted: row[name] = _rollup_value(fn, field, children) # ---- Joins --------------------------------------------------------------------------------------- # Another source's records joined onto each row (app/schemas/reporting.py # DatasetJoin) — a dataset with joins is a view over several sources. The # joined records are read through _prepare_query, so the viewer's own access # rules apply to them as to the rows themselves. # Joined records read per join at most. MAX_JOIN_ROWS = 50_000 def _check_join_aliases(src: record_sql.RecordSource, joins: list[dict] | None) -> set[str]: """The joins' aliases — each must be new: not a field or link of the source, and not used twice.""" aliases: set[str] = set() for j in joins or []: alias = j["alias"] if alias in src.declared or alias in aliases or alias in ("id", *_AUDIT_COLUMN_NAMES): from fastapi import HTTPException raise HTTPException(status_code=422, detail=f'Join name "{alias}" is already a field here or another join — pick another.') aliases.add(alias) return aliases def _join_key(value: Any) -> str: return value.strip().lower() if isinstance(value, str) else str(value) def _values(raw: Any) -> list: return [v for v in (raw if isinstance(raw, list) else [raw]) if v is not None and v != ""] async def _apply_joins(db: AsyncSession, tenant_id: uuid.UUID, rows: list[dict], joins: tuple | list, user_id: uuid.UUID | None) -> list[dict]: """Every join in order, then the joined link and person columns shown by name — after all of them, so a later join can match an earlier one's raw link ("emp.dept" → Departments' id).""" from fastapi import HTTPException labels: dict[str, tuple] = {} people: set[str] = set() for j in joins: alias, on, key_field = j["alias"], j["on_field"], j.get("join_field") or "id" columns = [c for c in j.get("columns") or [] if c] raw_keys = {v for r in rows for v in _values(r.get(on))} joined: list[dict] = [] if raw_keys: filters = list(j.get("filters") or []) if key_field == "id" and len(raw_keys) <= 500: # Just the records these rows point at. filters.append({"field": "id", "operator": "in", "value": sorted(str(v) for v in raw_keys)}) jquery = await _prepare_query(db, tenant_id, j["source_type"], uuid.UUID(str(j["source_id"])), filters, user_id) jquery.refuse_hidden(key_field, *columns) joined, total = await _fetch_and_filter_rows(db, tenant_id, jquery, set()) # raw — labelled below for c in columns: spec = _label_spec(jquery.src.field_def(c)) if spec: labels[f"{alias}.{c}"] = spec elif c in _AUDIT_USER_FIELDS and c not in jquery.src.declared: people.add(f"{alias}.{c}") if total > MAX_JOIN_ROWS: raise HTTPException(status_code=422, detail=f'Join "{alias}" matches {total} records — narrow it with filters (at most {MAX_JOIN_ROWS}).') index: dict[str, list[dict]] = {} for record in joined: for v in _values(record.get(key_field)): index.setdefault(_join_key(v), []).append(record) def merged(row: dict, record: dict | None) -> dict: out = {**row, f"{alias}.id": record.get("id") if record else None} for c in columns: out[f"{alias}.{c}"] = record.get(c) if record else None return out result: list[dict] = [] for row in rows: matches: list[dict] = [] seen: set[str] = set() for v in _values(row.get(on)): for record in index.get(_join_key(v), []): if record.get("id") not in seen: seen.add(record.get("id")) matches.append(record) if not matches: if j.get("type", "left") == "left": result.append(merged(row, None)) elif j.get("match", "first") == "all": result.extend(merged(row, m) for m in matches) else: result.append(merged(row, matches[0])) rows = result if labels: await _apply_reference_labels(db, tenant_id, rows, labels) if people: names = await _user_labels(db, tenant_id, {r.get(k) for r in rows for k in people} - {None}) for r in rows: for k in people: if r.get(k) is not None and str(r[k]) in names: r[k] = names[str(r[k])][0] or names[str(r[k])][1] or r[k] return rows async def _linked_field_defs(db: AsyncSession, tenant_id: uuid.UUID, source_id: str, is_form: bool) -> dict[str, dict]: if is_form: form = ( await db.execute(select(Form).where(Form.tenant_id == tenant_id, Form.form_id == uuid.UUID(source_id), Form.is_current.is_(True))) ).scalar_one_or_none() defs = form.field_definitions if form else [] else: master = (await db.execute(select(Master).where(Master.tenant_id == tenant_id, Master.id == uuid.UUID(source_id)))).scalar_one_or_none() defs = master.field_schema if master else [] return {f["name"]: f for f in defs if isinstance(f, dict) and f.get("name")} def _record_columns(src: record_sql.RecordSource) -> list: m = src.model return [m.id, m.field_values, m.created_at, m.updated_at, src.column("created_by"), m.updated_by] def _row_dict(declared: set[str], record_id: Any, field_values: dict, created_at: Any, updated_at: Any, created_by: Any, updated_by: Any) -> dict: # The record's own primary key isn't part of field_values (it's a real # column, not user-defined data) — merge it in under a synthetic "id" # key so it's selectable/toggleable like any other column (e.g. in the # Data Explorer's column picker), appended last so it never disturbs # the "first few columns" a dashboard widget shows when no explicit # column list was requested. A field_values row that happens to already # have a literal "id" key (unlikely, but JSONB has no reserved words) # is intentionally overridden here — the record's real primary key # should always win that name. created_at/updated_at/created_by/ # updated_by (Form's own created_by is named submitted_by on the model — # exposed under the shared name here for a uniform contract across both # source types) are merged in the same way — see _audit_columns — but # only for a name the schema itself hasn't already declared: a Master or # Form that (legacy data, before these were system-managed) defines its # own "created_by"-named field keeps showing that field's own stored # value rather than being silently overwritten by the system column. row = {**field_values, "id": str(record_id)} for k, v in _audit_columns(created_at, updated_at, created_by, updated_by).items(): if k not in declared: row[k] = v return row async def _decorate_rows( db: AsyncSession, tenant_id: uuid.UUID, src: record_sql.RecordSource, rows: list[dict], wanted_ref_fields: set[str] | None, labels: dict[str, tuple[str, str]] | None = None, ) -> None: """Shows linked records and people by name: "reference"-type fields get their linked record's display value, Created by / Updated by the user's name. `wanted_ref_fields` follows _resolve_reference_fields' convention — None means every such field, a set restricts it to those names.""" if src.master is not None: await _resolve_reference_fields(db, tenant_id, src.master, rows, wanted_ref_fields) elif src.form is not None: await _resolve_form_reference_fields(db, tenant_id, src.form, rows, wanted_ref_fields) declared = src.declared fields = [f for f in _AUDIT_USER_FIELDS if f not in declared and (wanted_ref_fields is None or f in wanted_ref_fields)] if not fields or not rows: return if labels is None: labels = await _user_labels(db, tenant_id, {r.get(f) for r in rows for f in fields}) for field in fields: for r in rows: uid = r.get(field) if uid is not None and str(uid) in labels: r[field] = labels[str(uid)][0] or labels[str(uid)][1] or uid async def _fetch_and_filter_rows( db: AsyncSession, tenant_id: uuid.UUID, query: _Query, wanted_ref_fields: set[str] | None, ) -> tuple[list[dict], int]: """Every row the query matches, filtered and decorated — the path for queries that need all rows in Python (custom expressions, a sort or aggregate over linked-record labels, a filter SQL can't match exactly). Returns (rows, total).""" src = query.src declared = src.declared stmt = select(*_record_columns(src)).where(*query.where).order_by(*src.default_order()) rows = [_row_dict(declared, *r) for r in await record_sql.read_capped(db, record_sql.with_tables(src, stmt))] if query.lookups: await _attach_lookups(db, tenant_id, src, rows, query.lookups) if query.rollups: await _attach_rollups(db, tenant_id, src, rows, query.rollups, query.user_id) if query.joins: rows = await _apply_joins(db, tenant_id, rows, query.joins, query.user_id) _compute_columns(rows, query.computed) labels: dict[str, tuple[str, str]] | None = None if not query.exact: if query.allowed_scope is not None and src.master is not None: rows = [r for r in rows if _record_scope_values(r, src.master.scope_field) & query.allowed_scope] # System Created by / Updated by columns (not a same-named declared field). audit_user_fields = {f for f in _AUDIT_USER_FIELDS if f not in declared} if any(f["field"] in audit_user_fields for f in conditions(query.filters)): labels = await _user_labels(db, tenant_id) for f in query.filters: rows = [r for r in rows if _row_passes(f, r, audit_user_fields, labels)] total = len(rows) await _decorate_rows(db, tenant_id, src, rows, wanted_ref_fields, labels) return query.strip(rows), total def _compute_columns(rows: list[dict], computed: tuple | list | None) -> None: """Each calculated column's value on each row, in order — so one can use an earlier one. A value that can't be worked out is None.""" if not computed: return from app.core.expr import evaluate for row in rows: for c in computed: row[c["name"]] = evaluate(c["expression"], row) async def _count(db: AsyncSession, tenant_id: uuid.UUID, query: _Query) -> int: stmt = select(func.count()).select_from(query.src.anchor).where(*query.where) return (await summary_cache.read(db, tenant_id, query.src, record_sql.with_tables(query.src, stmt)))[0][0] def _sorts_by_label(src: record_sql.RecordSource, field: str, wanted: set[str] | None) -> bool: """A linked field shown as its linked record's label sorts by that label, which only exists after the rows are read.""" return src.is_linked_field(field) and (wanted is None or field in wanted) def _aggregates_labels(src: record_sql.RecordSource, field: str | None) -> bool: """An aggregated field is always resolved for display first, so a linked field or Created by / Updated by aggregates over labels, not stored ids.""" return bool(field) and (src.is_linked_field(field) or src.field_kind(field) == "user") async def _sql_rows( db: AsyncSession, tenant_id: uuid.UUID, query: _Query, sort_field: str | None, sort_dir: str, columns: list[str], limit: int | None, wanted: set[str] | None, ) -> tuple[list[dict], int, list[str]]: src = query.src order = [] if sort_field: order = record_sql.sort_keys(src, sort_field, sort_dir == "desc", tenant_id, wanted is None or sort_field in wanted) stmt = select(*_record_columns(src)).where(*query.where).order_by(*order, *src.default_order()) declared = src.declared if limit is not None: fetched = (await db.execute(record_sql.with_tables(src, stmt.limit(limit)))).all() else: fetched = await record_sql.read_capped(db, record_sql.with_tables(src, stmt)) rows = [_row_dict(declared, *r) for r in fetched] total = await _count(db, tenant_id, query) if limit is not None and len(rows) >= limit else len(rows) await _decorate_rows(db, tenant_id, src, rows, wanted) rows = query.strip(rows) columns = [c for c in columns if c not in query.hidden] if columns: # The record's real id rides along even when it wasn't explicitly # requested (it never appears in column_names, so it doesn't grow # any rendered table — see PageBlockRenderer.tsx's PreviewDataList, # which needs the real id, not a fabricated one, to pass with # row-click navigation). return [{**{c: r.get(c) for c in columns}, "id": r.get("id")} for r in rows], total, columns column_names: list[str] = [] seen: set[str] = set() for r in rows: for k in r: if k not in seen and k not in _AUDIT_COLUMN_NAMES: seen.add(k) column_names.append(k) if total > len(rows): # Rows past this page may hold fields none of these do; the declared # ones are known without reading them. column_names += [n for n in (f["name"] for f in src.field_defs) if n not in seen and n not in _AUDIT_COLUMN_NAMES and n not in query.hidden] return rows, total, column_names async def _sql_group_rows( db: AsyncSession, tenant_id: uuid.UUID, query: _Query, group_by: str, aggregate_field: str | None, aggregate_fn: str, sort_field: str | None, sort_dir: str, limit: int | None, ) -> tuple[list[dict], int, list[str]]: """_aggregate_rows in SQL: one group per stored value, each with its row count and the aggregate field's sum/count/min/max. The stored values are then labelled exactly as rows are (linked records and people by name) and groups sharing a label merged — so the result matches the Python path while reading one row per distinct value.""" src = query.src stmt = ( select( record_sql.group_key(src, group_by), func.count(), *record_sql.number_partials(src, aggregate_field if aggregate_fn != "count" else None), ) .where(*query.where) .group_by(literal_column("1")) .order_by(func.min(src.order_column())) ) groups = await summary_cache.read(db, tenant_id, src, record_sql.with_tables(src, stmt.limit(record_sql.MAX_PYTHON_ROWS + 1))) record_sql.refuse_too_many(groups) keyed = [{group_by: record_sql.row_value(src, group_by, g[0])} for g in groups] await _decorate_rows(db, tenant_id, src, keyed, {group_by}) output_key = "count" if aggregate_fn == "count" else f"{aggregate_fn}_{aggregate_field}" counts: dict[str, int] = {} partials: dict[str, list | None] = {} for row, g in zip(keyed, groups): raw = row.get(group_by) label = "(empty)" if raw is None or raw == "" else str(raw) counts[label] = counts.get(label, 0) + g[1] partials[label] = record_sql.merge_partials(partials.get(label), tuple(g[2:6])) result = [ { group_by: label, output_key: counts[label] if aggregate_fn == "count" else record_sql.aggregate_from_partials(aggregate_fn, partials[label]), } for label in counts ] _sort_aggregated(result, group_by, output_key, sort_field, sort_dir) if limit is not None: result = result[:limit] return result, sum(counts.values()), [group_by, output_key] async def _query_source_rows( db: AsyncSession, tenant_id: uuid.UUID, source_id: uuid.UUID, filters: list[dict], user_id: uuid.UUID | None, computed: list[dict] | None = None, joins: list[dict] | None = None, ) -> list[dict]: """A Data Query's rows (app/crud/data_queries.py), filtered: an "equals" filter on one of its parameters is passed to the query instead.""" from fastapi import HTTPException from app.crud import data_queries query = await data_queries.load(db, tenant_id, source_id) if query is None: raise HTTPException(status_code=404, detail="Data query not found.") if not await data_queries.can_run(db, tenant_id, user_id, query): raise HTTPException(status_code=403, detail="Your role can't run this data query.") filters = await prepare_filters(db, tenant_id, user_id, filters) params, rest = data_queries.split_param_filters(query, filters) try: result = await data_queries.run(db, query, params) except data_queries.DataQueryError as exc: raise HTTPException(status_code=422, detail=f"{query.name}: {exc}") from exc rows = [dict(r) for r in result.rows] if joins: rows = await _apply_joins(db, tenant_id, rows, joins, user_id) _compute_columns(rows, computed) # A query's columns aren't typed: one whose values are numbers is numeric. numeric = {f["field"] for f in conditions(rest) if any(isinstance(r.get(f["field"]), (int, float)) and not isinstance(r.get(f["field"]), bool) for r in rows)} rest = coerce_filter_numbers(rest, numeric) return [r for r in rows if passes(rest, lambda f: _eval_operator(f["operator"], r.get(f["field"]), f.get("value")))] def _shape_rows( rows: list[dict], sort_field: str | None, sort_dir: str, columns: list[str], limit: int | None, group_by: str | None, aggregate_field: str | None, aggregate_fn: str | None, expression: str | None, ) -> tuple[list[dict], int, list[str]]: """resolve_rows' in-memory path for rows already fetched and filtered.""" total = len(rows) if expression: from app.crud.expr_eval import evaluate_expression return [{"value": evaluate_expression(expression, rows)}], total, ["value"] if group_by: rows, column_names = _aggregate_rows(rows, group_by, aggregate_field, aggregate_fn or "count", sort_field, sort_dir) return (rows[:limit] if limit is not None else rows), total, column_names if sort_field: rows.sort(key=lambda r: _sort_key(r.get(sort_field)), reverse=(sort_dir == "desc")) if columns: rows = [{c: r.get(c) for c in columns} for r in rows] column_names = list(columns) else: column_names = list(dict.fromkeys(k for r in rows for k in r)) return (rows[:limit] if limit is not None else rows), total, column_names async def resolve_rows( db: AsyncSession, tenant_id: uuid.UUID, source_type: str, source_id: uuid.UUID, filters: list[dict], sort_field: str | None, sort_dir: str, columns: list[str], limit: int | None = None, group_by: str | None = None, aggregate_field: str | None = None, aggregate_fn: str | None = None, user_id: uuid.UUID | None = None, expression: str | None = None, computed: list[dict] | None = None, joins: list[dict] | None = None, ) -> tuple[list[dict], int, list[str]]: """Returns (rows, total_before_limit, column_names). `computed` adds calculated columns ({name, expression}) to every row. Without group_by, column_names is `columns` when given, otherwise the union of keys actually present across the matched rows (in first-seen order) — lets the Data Explorer render a sensible table header for an ad-hoc "select all" query. With group_by, rows become one aggregated row per distinct group_by value and column_names is always [group_by, ] — see _aggregate_rows. `expression` (Phase 4 custom metrics) takes priority over group_by/aggregate_fn when set: it's evaluated once against the whole filtered row set and returned as a single synthetic row `[{"value": }]`, column_names `["value"]` — this is what a KPI widget's custom-expression config calls instead of the plain group_by/aggregate_fn path.""" # An empty `columns` means "every field" throughout this module (see the # column_names union-of-keys branch below) — so it must mean "resolve # every reference field" here too, not "resolve none": pass None (no # restriction) rather than an empty set. wanted: set[str] | None = None if columns: wanted = set(columns) if group_by: wanted.add(group_by) if aggregate_field: wanted.add(aggregate_field) elif group_by or aggregate_field: wanted = {f for f in (group_by, aggregate_field) if f} if source_type == "query": rows = await _query_source_rows(db, tenant_id, source_id, filters, user_id, computed, joins) return _shape_rows(rows, sort_field, sort_dir, columns, limit, group_by, aggregate_field, aggregate_fn, expression) query = await _prepare_query(db, tenant_id, source_type, source_id, filters, user_id, computed, joins) src = query.src query.refuse_hidden(sort_field, group_by, aggregate_field) query.use_lookups({c for c in (*columns, sort_field, group_by, aggregate_field) if _is_lookup(src, c)}) query.use_rollups({c for c in (*columns, sort_field, group_by, aggregate_field) if is_rollup(c)}) columns = [c for c in columns if c not in query.hidden] if expression: from app.crud.expr_eval import evaluate_expression rows, total = await _fetch_and_filter_rows(db, tenant_id, query, wanted) value = evaluate_expression(expression, rows) return [{"value": value}], total, ["value"] if group_by: fn = aggregate_fn or "count" if query.exact and not (fn != "count" and _aggregates_labels(src, aggregate_field)): return await _sql_group_rows(db, tenant_id, query, group_by, aggregate_field, fn, sort_field, sort_dir, limit) rows, total = await _fetch_and_filter_rows(db, tenant_id, query, wanted) rows, column_names = _aggregate_rows(rows, group_by, aggregate_field, fn, sort_field, sort_dir) if limit is not None: rows = rows[:limit] return rows, total, column_names if query.exact and not (sort_field and _sorts_by_label(src, sort_field, wanted)): return await _sql_rows(db, tenant_id, query, sort_field, sort_dir, columns, limit, wanted) rows, total = await _fetch_and_filter_rows(db, tenant_id, query, wanted) if sort_field: rows.sort(key=lambda r: _sort_key(r.get(sort_field)), reverse=(sort_dir == "desc")) if columns: rows = [{**{c: r.get(c) for c in columns}, "id": r.get("id")} for r in rows] column_names = columns else: column_names = [] seen = set() for r in rows: for k in r: if k not in seen and k not in _AUDIT_COLUMN_NAMES: seen.add(k) column_names.append(k) if limit is not None: rows = rows[:limit] return rows, total, column_names def _series_labels(series: list[dict]) -> list[str]: return [s.get("label") or s.get("expression") or f"{s.get('aggregate')}_{s.get('field')}" for s in series] async def _sql_grouped_series( db: AsyncSession, tenant_id: uuid.UUID, query: _Query, x_field: str, x_bucket: str, series: list[dict], limit: int | None, ) -> tuple[list[dict], list[str]]: """resolve_grouped_series in SQL — see _sql_group_rows. A date-bucketed X groups by calendar day in SQL (then by week/month here), so a chart of a million timestamps reads one row per day.""" src = query.src kind = src.field_kind(x_field) bucket_requested = bool(x_bucket and x_bucket != "none") by_day = bucket_requested and (kind == "time" or (kind == "json" and not src.is_linked_field(x_field))) if by_day: day = record_sql.day_key(src, x_field) # A value that isn't a date keeps its own group, for "(invalid)". keys = [day] if kind == "time" else [day, case((day.is_(None), record_sql.group_key(src, x_field)))] else: keys = [record_sql.group_key(src, x_field)] aggregated = [s for s in series if s.get("aggregate") != "count"] partial_cols = [col for s in aggregated for col in record_sql.number_partials(src, s.get("field"))] stmt = ( select(*keys, func.count(), *partial_cols) .where(*query.where) .group_by(*(literal_column(str(i + 1)) for i in range(len(keys)))) .order_by(func.min(src.order_column())) ) groups = await summary_cache.read(db, tenant_id, src, record_sql.with_tables(src, stmt.limit(record_sql.MAX_PYTHON_ROWS + 1))) record_sql.refuse_too_many(groups) keyed: list[dict] = [] for g in groups: if by_day and g[0] is not None: keyed.append({x_field: g[0]}) else: raw = g[1] if by_day and kind == "json" else (None if by_day else g[0]) keyed.append({x_field: record_sql.row_value(src, x_field, raw)}) await _decorate_rows(db, tenant_id, src, keyed, {x_field}) # Same rule as resolve_grouped_series: bucket only when a value is a date. bucketing = bucket_requested and any(isinstance(_coerce_date(r[x_field]), (datetime, date)) for r in keyed) counts: dict[str, int] = {} partials: dict[str, list[list | None]] = {} for row, g in zip(keyed, groups): raw = row[x_field] label = _bucket_date(raw, x_bucket) if bucketing else ("(empty)" if raw is None or raw == "" else str(raw)) counts[label] = counts.get(label, 0) + g[len(keys)] merged = partials.setdefault(label, [None] * len(aggregated)) base = len(keys) + 1 for i in range(len(aggregated)): merged[i] = record_sql.merge_partials(merged[i], tuple(g[base + 4 * i : base + 4 * i + 4])) order = list(counts) if bucketing: order.sort() labels = _series_labels(series) result: list[dict] = [] for x_label in order: out: dict[str, Any] = {"x": x_label} agg_index = 0 for s, label in zip(series, labels): if s.get("aggregate") == "count": out[label] = counts[x_label] else: out[label] = record_sql.aggregate_from_partials(s.get("aggregate") or "count", partials[x_label][agg_index]) agg_index += 1 result.append(out) if limit is not None: result = result[:limit] return result, labels async def resolve_grouped_series( db: AsyncSession, tenant_id: uuid.UUID, source_type: str, source_id: uuid.UUID, filters: list[dict], x_field: str, x_bucket: str, series: list[dict], limit: int | None = None, user_id: uuid.UUID | None = None, computed: list[dict] | None = None, joins: list[dict] | None = None, ) -> tuple[list[dict], list[str]]: """The chart-widget counterpart to resolve_rows' group_by — supports multiple Y series (each a field+aggregate or a custom expression, see schemas/navigation.py::WidgetSeriesDef) grouped by one X field, with optional date bucketing. Returns (rows, series_labels): each row is `{"x": , "": , ...}` — one row per distinct x value, in first-seen order — which a Recharts / consumes directly via dataKey={label}.""" if source_type == "query": rows = await _query_source_rows(db, tenant_id, source_id, filters, user_id, computed, joins) return _series_from_rows(rows, x_field, x_bucket, series, limit) query = await _prepare_query(db, tenant_id, source_type, source_id, filters, user_id, computed, joins) src = query.src query.refuse_hidden(x_field, *(s.get("field") for s in series)) query.use_lookups({c for c in (x_field, *(s.get("field") for s in series)) if _is_lookup(src, c)}) query.use_rollups({c for c in (x_field, *(s.get("field") for s in series)) if is_rollup(c)}) if query.exact and not any( s.get("expression") or (s.get("aggregate") != "count" and _aggregates_labels(src, s.get("field"))) for s in series ): return await _sql_grouped_series(db, tenant_id, query, x_field, x_bucket, series, limit) wanted = {x_field} | {s["field"] for s in series if s.get("field")} rows, _ = await _fetch_and_filter_rows(db, tenant_id, query, wanted) return _series_from_rows(rows, x_field, x_bucket, series, limit) def _series_from_rows(rows: list[dict], x_field: str, x_bucket: str, series: list[dict], limit: int | None) -> tuple[list[dict], list[str]]: from app.crud.expr_eval import evaluate_expression # Date bucketing only makes sense on a date-valued X field. A widget can be # left with a bucket (the designer defaults one) while its X field is a # category — an office, a status — whose values would all land in # "(invalid)" and collapse the chart into one bar. So the bucket applies # only when at least one X value actually parses as a date; otherwise the # field is grouped by its own values, as if the bucket were "none". bucketing = bool(x_bucket and x_bucket != "none") and any( isinstance(_coerce_date(r.get(x_field)), (datetime, date)) for r in rows ) groups: dict[str, list[dict]] = {} order: list[str] = [] for r in rows: raw = r.get(x_field) label = _bucket_date(raw, x_bucket) if bucketing else ( "(empty)" if raw is None or raw == "" else str(raw) ) if label not in groups: groups[label] = [] order.append(label) groups[label].append(r) if bucketing: order.sort() labels = _series_labels(series) result: list[dict] = [] for x_label in order: group_rows = groups[x_label] out: dict[str, Any] = {"x": x_label} for s, label in zip(series, labels): if s.get("expression"): out[label] = evaluate_expression(s["expression"], group_rows) elif s.get("aggregate") == "count": out[label] = len(group_rows) else: out[label] = _aggregate_value(s.get("aggregate") or "count", [r.get(s.get("field")) for r in group_rows]) result.append(out) if limit is not None: result = result[:limit] return result, labels