Response Serializers

Declarative DB row-to-dict conversion using pytastic response schemas. Replaces manual row_to_*() dict-building with typed, validated schemas.

Module: utils/serializers.py Depends on: pytastic (setter, getter, default=)

Quick start

from utils.serializers import sx, serialize, create

# 1. Define a response schema in your app's schema.py
class TaskOut(TypedDict):
    id: str
    title: str
    description: Annotated[str, "default="]
    labels: Annotated[list, "setter=json_or_list; default=[]"]
    completed: Annotated[bool, "setter=to_bool; default=false"]
    due_date: NotRequired[Annotated[str, "setter=date_to_str"]]
    size_label: NotRequired[Annotated[str, "factory=task_size_label"]]

# 2. Register any app-specific factories
sx.factory("task_size_label", lambda d: TASK_SIZES.get(d.get("size")) if d.get("size") else None)

# 3. Convert DB rows
task_dict = serialize(TaskOut, db_row)          # plain dict (for API envelopes)
task_dot  = create(TaskOut, db_row)             # DotDict (for templates)

Two entry points

Both call row.data() to extract a raw dict from the supersql Result, merge any **extra overrides, then validate through the schema with strip=True (extra columns like workspace_id are removed).

DotDict is a dict subclass — it passes isinstance(d, dict) and works anywhere a plain dict works. Nested dicts are lazy-wrapped for deep dot access (task.member.name).

Schema design rules

Use default= for nullable columns that need a fallback

default= fills when the key is missing or the value is None. This covers both absent keys and NULL DB columns.

description: Annotated[str, "default="]           # None/missing → ""
priority: Annotated[str, "default=medium"]         # None/missing → "medium"
position: Annotated[int, "default=0"]              # None/missing → 0
labels: Annotated[list, "default=[]"]              # None/missing → []
completed: Annotated[bool, "default=false"]        # None/missing → False

Use setter= for type transformations

Setters transform a value before validation. Only needed when the DB stores a different type than the response expects.

Combine setter + default when both a transformation and a fallback are needed:

labels: Annotated[list, "setter=json_or_list; default=[]"]
completed: Annotated[bool, "setter=to_bool; default=false"]

Execution order: setter runs first, then default fills if the setter returned None, then type validation.

Use factory= for computed fields

Factories compute a value from the full input dict. The field doesn't need to exist in the DB row.

# Register once at module level
sx.factory("task_size_label", lambda d: TASK_SIZES.get(d.get("size")) if d.get("size") else None)

# Use in schema
size_label: NotRequired[Annotated[str, "factory=task_size_label"]]

Factories see the original input dict (before setters run on other fields).

Pytastic 0.5.0 renamed the old "getter" (full-dict, pre-validation) to factory. The new getter= runs after validation on a single field's validated value — useful for hydrating a primitive into a richer Python type (e.g. ISO string → date). It is registered with vx.getter(name, fn) on the inbound app._pytastic instance, not on sx.

Use NotRequired for truly optional fields

Fields that may or may not be in the output. If the DB column can be NULL and None is an acceptable output value (not needing a default), use NotRequired:

due_date: NotRequired[Annotated[str, "setter=date_to_str"]]  # None → omitted from output
schedule_id: NotRequired[str]                                  # None → omitted from output

Handling member lookups

Pytastic getters/setters are sync, but member resolution requires async DB queries. Keep member resolution as a manual post-step:

def row_to_worker(row, members: dict[str, dict]) -> dict:
    worker = serialize(WorkerOut, row)
    worker["member"] = lookup_member(members, worker["member_mrn"])
    worker["employment_type_label"] = EMPLOYMENT_TYPE_LABELS.get(worker["employment_type"], worker["employment_type"])
    return worker

The schema handles all the column-level transformations; the function adds cross-table lookups.

Registering custom setters and factories

Register on the shared sx instance at module level (never inside functions):

# In utils/serializers.py (shared across all apps):
sx.setter("json_or_list", _json_or_list)

# In app helpers (app-specific):
from utils.serializers import sx
sx.factory("task_size_label", lambda d: ...)

Registration happens once at import time. The name in sx.factory("name", fn) must match the factory=name in the schema annotation.

Response schemas live in schema.py

Each app's schema.py contains both input schemas (for request validation) and output schemas (for response serialization). Convention:

• Input schemas: verb-based names — CreateTask, UpdateWorker
• Output schemas: noun + Out suffix — TaskOut, WorkerOut, BoardOut

apps/tasks/schema.py
├── CreateTask          ← input (expects= on route)
├── UpdateTask          ← input
├── TaskOut             ← output (serialize/create)
├── BoardOut            ← output
└── ScheduleOut         ← output

Relationship to Heaven's returns=

Heaven's app.schema.METHOD(route, returns=Schema) registers an AFTER hook that validates res.body against the schema. This works for direct responses where res.body is the raw data.

Current API handlers wrap responses in one()/many() envelopes, so returns= would validate the envelope structure — not the individual items. Use serialize() in helpers for now; returns= is available for endpoints that set res.body directly and for OpenAPI doc generation.

Full example

# apps/workers/schema.py
class CompensationOut(TypedDict):
    id: str
    worker_id: str
    effective_date: NotRequired[Annotated[str, "setter=date_to_str"]]
    items: Annotated[list, "setter=json_or_list; default=[]"]
    total_amount: Annotated[float, "setter=to_float; default=0"]
    amount: Annotated[float, "setter=to_float; default=0"]
    currency: Annotated[str, "default=USD"]
    pay_frequency: Annotated[str, "default=monthly"]
    reason: Annotated[str, "default="]
    notes: Annotated[str, "default="]
    created_at: NotRequired[Annotated[str, "setter=date_to_str"]]

# apps/workers/helpers.py
def row_to_compensation(row) -> dict:
    return serialize(CompensationOut, row)

Before (manual):

def row_to_compensation(row) -> dict:
    items = row.column("items") or []
    if isinstance(items, str):
        items = json.loads(items)
    return {
        "id": row.id,
        "worker_id": row.worker_id,
        "effective_date": str(row.effective_date) if row.effective_date else None,
        "items": items,
        "total_amount": float(row.total_amount) if row.total_amount else 0,
        "amount": float(row.amount) if row.amount else 0,
        "currency": row.currency or "USD",
        "pay_frequency": row.pay_frequency or "monthly",
        "reason": row.reason or "",
        "notes": row.notes or "",
        "created_at": str(row.created_at) if row.created_at else None,
    }