"""Public (signed-out) read access to a datasource's records or a form's submissions — what the public website's data blocks show (Data List, Detail View, Label, KPI card, charts). Off unless turned on per source (Datasources / Forms › Public access, a PublicDataSource row), and then: * only the chosen `fields` are ever returned, filtered, sorted, grouped or searched on — plus the record's id (a Detail View finds its record by it). A request naming any other field is refused, so a hidden value can't be probed with a filter either. Lookups, rollups, joins, calculated columns and custom expressions aren't available publicly; * every read is narrowed by the source's fixed `filters` (e.g. status = Published), on top of the block's own; * a form's drafts are never included, and the source's record access rules for everyone still apply — a visitor has no roles; * values read as a visitor sees them: a linked record, a person, or a datasource-backed choice by its name; files and signatures aren't offered at all. Turning access off (or unticking a field) takes effect on the next read. Public pages may hold data blocks only over such sources and only read-only (data_block_problems).""" import uuid from typing import Any from fastapi import HTTPException from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from app.crud.filter_tree import conditions from app.models.master import Master from app.models.public_data import PublicDataSource from app.models.workflow import Form # Never offered publicly: files (nothing to serve them by), signatures, # passwords, and values that only make sense with a session. _MASTER_NOT_PUBLIC = {"file"} _FORM_NOT_PUBLIC = {"file_upload", "image_upload", "attachment", "multi_attachment", "camera", "signature", "password", "embedded_form", "button"} _FORM_NO_VALUE = {"section", "panel", "tabs", "accordion", "wizard", "row", "column", "group_end", "button"} # Searchable when shown: plain text a visitor would type part of. _MASTER_TEXT = {"text", "varchar", "char", "single_select"} _FORM_TEXT = {"text", "textarea", "email", "url", "dropdown", "radio", "combobox", "autocomplete", "phone", "barcode", "autonumber", "label", "markdown"} AUDIT_FIELDS = ("created_at", "updated_at") DATA_BLOCK_TYPES = {"data_list", "detail_view", "label", "kpi_card", "bar_chart", "pie_chart", "line_chart"} _PUBLIC_ROW_ACTIONS = {"navigate", "open_external_url", "open_popup", "print"} class PublicDataError(Exception): def __init__(self, message: str, status_code: int = 422): super().__init__(message) self.status_code = status_code # ---- The source and its public fields ------------------------------------------------------------ async def source_fields(db: AsyncSession, tenant_id: uuid.UUID, source_type: str, source_id: uuid.UUID) -> tuple[str, list[dict]] | None: """(name, field definitions) of a datasource or a form's current version.""" if source_type == "master": master = (await db.execute(select(Master).where(Master.tenant_id == tenant_id, Master.id == source_id))).scalar_one_or_none() return (master.name, list(master.field_schema or [])) if master else None form = (await db.execute(select(Form).where(Form.tenant_id == tenant_id, Form.form_id == source_id, Form.is_current.is_(True)))).scalar_one_or_none() return (form.name, [f for f in form.field_definitions or [] if f.get("type") not in _FORM_NO_VALUE]) if form else None def offerable(source_type: str, field: dict) -> bool: """Whether a field can be shown to visitors at all.""" kind = field.get("type") return kind not in (_MASTER_NOT_PUBLIC if source_type == "master" else _FORM_NOT_PUBLIC) def field_label(source_type: str, field: dict) -> str: if source_type == "master": return (field.get("alias") or "").strip() or field.get("label") or field["name"].replace("_", " ") return field.get("label") or field["name"] def public_type(source_type: str, field: dict) -> str: """The field's type as a visitor sees it — a link or a person is text.""" kind = field.get("type") or "text" if source_type == "master": return "varchar" if kind in ("reference", "user") else kind if kind == "user" or field.get("master_id"): return "text" return kind async def get_config(db: AsyncSession, tenant_id: uuid.UUID, source_type: str, source_id: uuid.UUID) -> PublicDataSource | None: return ( await db.execute( select(PublicDataSource).where( PublicDataSource.tenant_id == tenant_id, PublicDataSource.source_type == source_type, PublicDataSource.source_id == source_id ) ) ).scalar_one_or_none() async def public_sources(db: AsyncSession, tenant_id: uuid.UUID) -> set[str]: """"master:" / "form:" of every source open to the public.""" rows = await db.execute( select(PublicDataSource.source_type, PublicDataSource.source_id).where(PublicDataSource.tenant_id == tenant_id, PublicDataSource.enabled.is_(True)) ) return {f"{kind}:{sid}" for kind, sid in rows.all()} def check_config(source_type: str, defs: list[dict], fields: list[str], filters: list[dict]) -> list[str]: """Problems with a config, one sentence each.""" by_name = {f["name"]: f for f in defs} problems = [] for name in fields: if name in AUDIT_FIELDS: continue f = by_name.get(name) if f is None: problems.append(f"“{name}” isn't a field of this {'datasource' if source_type == 'master' else 'form'}.") elif not offerable(source_type, f): problems.append(f"“{field_label(source_type, f)}” can't be shown publicly (files, signatures and passwords never are).") known = set(by_name) | set(AUDIT_FIELDS) | {"id", "created_by", "updated_by"} for c in conditions(filters): if c.get("field") not in known: problems.append(f"The filter on “{c.get('field')}” names a field this source doesn't have.") return problems async def open_source(db: AsyncSession, tenant_id: uuid.UUID, source_type: str, source_id: uuid.UUID) -> tuple[PublicDataSource, str, list[dict]]: """The source's public config, name and the definitions of its public fields — or PublicDataError 404 when it isn't open to the public.""" config = await get_config(db, tenant_id, source_type, source_id) found = await source_fields(db, tenant_id, source_type, source_id) if config and config.enabled else None if config is None or not config.enabled or found is None: raise PublicDataError("This data isn't public.", 404) name, defs = found by_name = {f["name"]: f for f in defs} shown = [] for n in config.fields or []: if n in AUDIT_FIELDS: shown.append({"name": n, "type": "datetime", "label": "Created" if n == "created_at" else "Updated"}) elif n in by_name and offerable(source_type, by_name[n]): shown.append(by_name[n]) return config, name, shown # ---- Reading ------------------------------------------------------------------------------------ def _require_public(fields: set[str], names: list[str | None], what: str) -> None: bad = sorted({n for n in names if n and n != "id" and n not in fields}) if bad: raise PublicDataError(f"{what} can only use the public fields — not {', '.join(bad)}.", 403) def _search_group(source_type: str, shown: list[dict], q: str) -> dict | None: text = _MASTER_TEXT if source_type == "master" else _FORM_TEXT names = [f["name"] for f in shown if f.get("type") in text] if not names: return None return {"match": "any", "filters": [{"field": n, "operator": "contains", "value": q} for n in names]} async def _name_people(db: AsyncSession, tenant_id: uuid.UUID, source_type: str, shown: list[dict], rows: list[dict]) -> None: """A person field reads as the person's name.""" from app.crud.dataset_query import _user_labels people = [f["name"] for f in shown if f.get("type") == "user"] if not people or not rows: return labels = await _user_labels(db, tenant_id, {str(r.get(p)) for r in rows for p in people if r.get(p)}) for r in rows: for p in people: uid = r.get(p) if uid and str(uid) in labels: r[p] = labels[str(uid)][0] or labels[str(uid)][1] async def query(db: AsyncSession, tenant_id: uuid.UUID, payload: dict) -> tuple[list[dict], int, list[str]]: """A data block's read: (rows, total, columns), limited to the public fields and narrowed by the source's fixed filters.""" from app.crud.dataset_query import resolve_rows source_type, source_id = payload["source_type"], uuid.UUID(str(payload["source_id"])) config, _name, shown = await open_source(db, tenant_id, source_type, source_id) fields = {f["name"] for f in shown} filters = list(payload.get("filters") or []) _require_public(fields, [*payload.get("columns", []), payload.get("sort_field"), payload.get("group_by"), payload.get("aggregate_field")], "A public page") _require_public(fields, [c.get("field") for c in conditions(filters)], "A filter on a public page") q = (payload.get("q") or "").strip() if q: if not config.allow_search: raise PublicDataError("Searching this data isn't allowed.", 403) group = _search_group(source_type, shown, q) filters.append(group or {"field": "id", "operator": "equals", "value": "__no_match__"}) columns = [c for c in payload.get("columns") or [] if c != "id"] or [f["name"] for f in shown] skip, limit = payload.get("skip", 0), payload.get("limit", 50) rows, total, column_names = await resolve_rows( db, tenant_id, source_type, source_id, [*config.filters, *filters], payload.get("sort_field"), payload.get("sort_dir") or "asc", columns, limit=skip + limit, group_by=payload.get("group_by"), aggregate_field=payload.get("aggregate_field"), aggregate_fn=payload.get("aggregate_fn"), user_id=None, ) rows = rows[skip:] if not payload.get("group_by"): keep = fields | {"id"} rows = [{k: v for k, v in r.items() if k in keep} for r in rows] column_names = [c for c in column_names if c in fields] await _name_people(db, tenant_id, source_type, shown, rows) return rows, total, column_names async def series(db: AsyncSession, tenant_id: uuid.UUID, payload: dict) -> tuple[list[dict], list[str]]: from app.crud.dataset_query import resolve_grouped_series source_type, source_id = payload["source_type"], uuid.UUID(str(payload["source_id"])) config, _name, shown = await open_source(db, tenant_id, source_type, source_id) fields = {f["name"] for f in shown} filters = list(payload.get("filters") or []) _require_public(fields, [payload.get("x_field"), *(s.get("field") for s in payload.get("series") or [])], "A chart on a public page") _require_public(fields, [c.get("field") for c in conditions(filters)], "A filter on a public page") return await resolve_grouped_series( db, tenant_id, source_type, source_id, [*config.filters, *filters], payload["x_field"], payload.get("x_bucket") or "none", [{"field": s.get("field"), "aggregate": s.get("aggregate"), "label": s.get("label")} for s in payload.get("series") or []], limit=payload.get("limit", 200), user_id=None, ) # ---- Public pages ------------------------------------------------------------------------------- def data_block_problems(block: dict, open_sources: set[str]) -> list[str]: """Why a data block can't be on a public page — empty when it can: it reads a source open to the public, and only reads.""" kind = block.get("type") or "" what = (block.get("label") or kind.replace("_", " ")).strip() source = block.get("data_source") or "" if kind == "label" and not source: return [] if source not in open_sources: return [ f"“{what}” shows data from a datasource or form that isn't public — turn on its Public access first (Datasources or Forms › Public access)." if source.split(":")[0] in ("master", "form") else f"“{what}” reads a saved dataset or data query, which can't be public — use a datasource or form with Public access instead." ] problems = [] if any(block.get(k) for k in ("allow_create", "allow_edit", "allow_delete", "allow_import", "allow_export", "inline_edit", "allow_saved_views")) or block.get("bulk_actions"): problems.append(f"“{what}” lets people add, change, import or export records — on a public page data can only be read.") actions = [block.get("row_click_action"), *((ra or {}).get("action") for ra in block.get("row_actions") or [])] for action in actions: if isinstance(action, dict) and action.get("type") and action["type"] not in _PUBLIC_ROW_ACTIONS: problems.append(f"“{what}” runs “{action['type'].replace('_', ' ')}” on a row, which needs a signed-in person.") elif isinstance(action, dict) and action.get("type") == "navigate" and action.get("target_type") not in (None, "page"): problems.append(f"“{what}” opens a {action.get('target_type')} from a row, which needs a signed-in person — link to a public page instead.") return problems def presented(source_type: str, field: dict) -> dict[str, Any]: """A public field as the public schema lists it.""" out = {"name": field["name"], "label": field_label(source_type, field), "type": public_type(source_type, field)} if field.get("options") and not field.get("master_id"): out["options"] = list(field["options"]) return out def http_error(exc: PublicDataError) -> HTTPException: return HTTPException(status_code=exc.status_code, detail=str(exc))