"""Datasource field rules that hold however a record is written — the record editor, a bulk import, the Super Admin editor, a task's data step: * default_value — filled in on create when the field is left blank: a fixed value, or {{today}} / {{now}} / {{current_user.id|name|email}}; * unique — no two live records share the value (checked under a lock, so two saves at once can't both get through); * autonumber fields — given the next number once, on create, in their format ("APP-{YYYY}-{SEQ:5}"), and never changed after. """ import re import uuid from datetime import datetime from typing import Any from sqlalchemy import func, select, text from sqlalchemy.ext.asyncio import AsyncSession from app.models.master import Master, MasterRecord _NUMBER_TYPES = {"number", "smallint", "integer", "bigint", "numeric", "real", "double_precision"} _WHOLE_TYPES = {"smallint", "integer", "bigint"} _TOKEN = re.compile(r"\{\{\s*([a-z_.]+)\s*\}\}") _FORMAT_TOKEN = re.compile(r"\{(SEQ(?::(\d{1,2}))?|YYYY|YY|MM|DD)\}") def _blank(v: Any) -> bool: return v is None or v == "" or v == [] # ---- defaults ------------------------------------------------------------------------------------ def _typed_default(field: dict, raw: Any) -> Any: """The default as the field's type stores it ("5" -> 5 on a number).""" ftype = field.get("type") if not isinstance(raw, str): return raw if ftype in _NUMBER_TYPES: try: n = float(raw) except ValueError: return raw return int(n) if ftype in _WHOLE_TYPES or n.is_integer() else n if ftype == "boolean": return raw.strip().lower() in ("true", "yes", "1", "on") return raw def resolve_default(field: dict, now: datetime, user: Any | None) -> Any: raw = field.get("default_value") if raw is None or raw == "": return None def token(m: re.Match) -> str: name = m.group(1) if name == "today": return now.date().isoformat() if name == "now": return now.isoformat() if name.startswith("current_user.") and user is not None: value = getattr(user, name.removeprefix("current_user."), None) return "" if value is None else str(value) return "" value = _TOKEN.sub(token, str(raw)).strip() return _typed_default(field, value) if value != "" else None def apply_defaults(field_schema: list[dict], values: dict, now: datetime, user: Any | None = None) -> dict: """`values` with each blank field that has a default filled in.""" out = dict(values) for field in field_schema: if field.get("type") in ("formula", "autonumber") or not _blank(out.get(field["name"])): continue default = resolve_default(field, now, user) if default is not None: out[field["name"]] = default return out async def _now_and_user(db: AsyncSession, tenant_id: uuid.UUID, user_id: uuid.UUID | None) -> tuple[datetime, Any]: from app.core.timezones import now_in, tenant_zone from app.models.identity import User user = await db.get(User, user_id) if user_id else None return now_in(await tenant_zone(db, tenant_id)), user # ---- auto-numbers -------------------------------------------------------------------------------- def format_autonumber(fmt: str | None, number: int, now: datetime) -> str: def token(m: re.Match) -> str: part = m.group(1) if part.startswith("SEQ"): width = int(m.group(2) or 0) return str(number).zfill(width) return {"YYYY": f"{now.year:04d}", "YY": f"{now.year % 100:02d}", "MM": f"{now.month:02d}", "DD": f"{now.day:02d}"}[part] return _FORMAT_TOKEN.sub(token, fmt or "{SEQ}") def _counter_key(master_id: uuid.UUID, field: dict, now: datetime, prefix: str = "autonumber") -> str: reset = field.get("autonumber_reset") or "never" period = f":{now.year:04d}" if reset == "yearly" else f":{now.year:04d}-{now.month:02d}" if reset == "monthly" else "" return f"{prefix}:{master_id}:{field['name']}{period}" async def assign_form_autonumbers(db: AsyncSession, tenant_id: uuid.UUID, form_id: uuid.UUID, field_defs: list[dict], values: dict, previous: dict | None) -> dict: """A form's autonumber fields: a number already given is kept; a missing one is given now (callers do this on submit, never for a draft). Counters are per form and field ("autonumber:form::").""" from app.core.timezones import now_in, tenant_zone from app.crud.sequence import next_sequence_value out = dict(values) now: datetime | None = None for field in field_defs: if field.get("type") != "autonumber": continue kept = (previous or {}).get(field["name"]) if not _blank(kept): out[field["name"]] = kept continue now = now or now_in(await tenant_zone(db, tenant_id)) number = await next_sequence_value(db, tenant_id, _counter_key(form_id, field, now, "autonumber:form")) out[field["name"]] = format_autonumber(field.get("autonumber_format"), number, now) return out async def assign_autonumbers(db: AsyncSession, tenant_id: uuid.UUID, master: Master, values: dict, now: datetime) -> dict: from app.crud.sequence import next_sequence_value out = dict(values) for field in master.field_schema: if field.get("type") == "autonumber": number = await next_sequence_value(db, tenant_id, _counter_key(master.id, field, now)) out[field["name"]] = format_autonumber(field.get("autonumber_format"), number, now) return out def keep_autonumbers(master: Master, values: dict, stored: dict) -> dict: """An update leaves auto-numbers as they were given.""" out = dict(values) for field in master.field_schema: if field.get("type") == "autonumber": if field["name"] in stored: out[field["name"]] = stored[field["name"]] else: out.pop(field["name"], None) return out # ---- unique -------------------------------------------------------------------------------------- def _unique_key(value: Any) -> str: # Text compares without case or surrounding spaces: "ABC " and "abc" clash. return value.strip().lower() if isinstance(value, str) else str(value) async def unique_errors( db: AsyncSession, tenant_id: uuid.UUID, master: Master, values: dict, record_id: uuid.UUID | None = None, seen: dict[str, set[str]] | None = None ) -> list[str]: """A message per unique field whose value another live record already has. `seen` collects values across one import, so two rows of the same file clash too. Takes a lock per field until the transaction ends, so a concurrent save of the same value waits and then sees this one.""" errors: list[str] = [] for field in master.field_schema: name = field["name"] value = values.get(name) if not field.get("unique") or _blank(value): continue key = _unique_key(value) label = field.get("alias") or name if seen is not None: if key in seen.setdefault(name, set()): errors.append(f"'{label}' must be unique — {value!r} appears more than once in this import") continue seen[name].add(key) await db.execute(text("SELECT pg_advisory_xact_lock(hashtext(:k))"), {"k": f"unique:{master.id}:{name}"}) stored = MasterRecord.field_values.op("->>")(name) stmt = select(MasterRecord.id).where( MasterRecord.tenant_id == tenant_id, MasterRecord.master_id == master.id, MasterRecord.is_deleted.is_(False), (func.lower(func.trim(stored)) == key) if isinstance(value, str) else (stored == str(value)), ) if record_id is not None: stmt = stmt.where(MasterRecord.id != record_id) if (await db.execute(stmt.limit(1))).scalar_one_or_none() is not None: errors.append(f"'{label}' must be unique — another record already has {value!r}") return errors # ---- the write paths ----------------------------------------------------------------------------- async def prepare_create(db: AsyncSession, tenant_id: uuid.UUID, user_id: uuid.UUID | None, master: Master, values: dict) -> dict: """Defaults filled in, before validation (a default can satisfy "required").""" now, user = await _now_and_user(db, tenant_id, user_id) return apply_defaults(master.field_schema, values, now, user) async def finish_create(db: AsyncSession, tenant_id: uuid.UUID, master: Master, values: dict) -> dict: """Auto-numbers given out — after validation, so a refused record uses no number.""" if not any(f.get("type") == "autonumber" for f in master.field_schema): return values from app.core.timezones import now_in, tenant_zone return await assign_autonumbers(db, tenant_id, master, values, now_in(await tenant_zone(db, tenant_id)))