Procedures

Some database work is neither a document write nor an analytical read: recomputing a rollup over a freshly-ingested batch, calling a stored function, refreshing a materialized view. You can always reach for the raw client — but then you own correctness, with no parameter validation, no tenancy, no portability. The procedures port is the governed middle ground: a spec-named, parametrized command or compute step. It is the write-and-compute twin of the governed, parametrized reads you run through analytics, sharing the same machinery and differing only in intent.

Where it sits¶

A handler touches the database three ways, and procedures fills the gap between the other two:

Procedures promotes raw SQL to a named, validated, tenant-aware, portable operation. Unlike analytics it mutates or computes rather than returning pages — so it is command-only, and acquiring it inside a read-only operation fails closed. The motivating case is a batch pipeline: ingest a large batch, then run one set-based statement over it, instead of per-row triggers that overload the database.

The shape in code¶

One spec is one procedure: a typed params model in, and an optional typed result. It names nothing about SQL or the backend — that lives in the wiring.

class Sale(BaseModel):
    region: str
    amount: int


class RegionTotal(BaseModel):
    region: str
    total: int


class RegionQuery(BaseModel):
    pass


class RecomputeWindow(BaseModel):
    since: str = "2026-01-01"


# The batch lands via analytics append; the read side serves recomputed per-region totals.
SALES_SPEC = AnalyticsSpec[RegionTotal, Sale](
    name="sales",
    read=RegionTotal,
    queries={"by_region": AnalyticsQueryDefinition(params=RegionQuery)},
    ingest=Sale,
)

# The procedure is the whole handler-facing surface: typed params, no result model means
# side-effect only (returns an affected-row count). Command-only — it mutates/computes.
RECOMPUTE_SPEC = ProcedureSpec[RecomputeWindow, None](
    name="recompute_region_totals",
    params=RecomputeWindow,
)

A handler resolves the port off the context and runs it with typed params. The recompute below replaces a swarm of per-row triggers with a single governed call:

async def ingest_sales(ctx: ExecutionContext, sales: list[Sale]) -> None:
    # Append-only ingest of the batch (stands in for a large bulk load).
    await ctx.analytics.ingest(SALES_SPEC).append(sales)


async def recompute(ctx: ExecutionContext, since: str = "2026-01-01") -> int:
    # ONE governed set-based statement over the whole batch — no per-row triggers.
    result = await ctx.procedure.command(RECOMPUTE_SPEC).run(
        RecomputeWindow(since=since)
    )
    return result.affected_count or 0


async def region_totals(ctx: ExecutionContext) -> list[RegionTotal]:
    page = await ctx.analytics.query(SALES_SPEC).run("by_region", RegionQuery())
    return list(page.hits)

What comes back¶

The result you declare on the spec drives the shape of the ExecResult. The cardinality is deliberately narrow — a scalar, a single row, or an affected-row count. Pages of rows are analytics' job, not this port's.

Mapping it to Postgres¶

The handler-facing spec stays backend-agnostic; the statement lives in the integration config. On Postgres, one PostgresProcedureConfig maps a route to a statement with %(name)s placeholders:

PostgresProcedureConfig(
    # A set-based statement; with result=None the rowcount is the affected count, so this is DML
    # (not `SELECT a_function(...)`, which returns one row — a count-returning function uses a
    # scalar result instead).
    sql=(
        "INSERT INTO region_totals (region, total) "
        "SELECT region, sum(amount) FROM sales WHERE since >= %(since)s GROUP BY region "
        "ON CONFLICT (region) DO UPDATE SET total = excluded.total"
    ),
    in_transaction=True,            # False for REFRESH MATERIALIZED VIEW CONCURRENTLY
    statement_timeout=None,         # SET LOCAL statement_timeout for long compute
    query_schema=None,              # per-tenant schema (namespace tier)
)

Set in_transaction=False for statements that cannot run inside a transaction (REFRESH MATERIALIZED VIEW CONCURRENTLY and some maintenance); they take the autocommit path. statement_timeout and query_schema both need a transaction, so they are rejected at wiring when combined with in_transaction=False.

Tenancy¶

Analytics and document writes can enforce tenant isolation because the framework composes the predicate. A procedure runs author-supplied opaque SQL, so the framework can only enforce isolation where the tenant boundary lives outside that SQL:

At the tagged tier the framework binds the current tenant id as the %(tenant)s parameter, but it cannot prove the SQL uses it to scope. So a tenant_aware route is allowed, but checked.

A procedure that is inherently cross-tenant — refreshing a shared materialized view, a cross-tenant rollup — is simply not tenant_aware, the same as any other non-tenant port.

The batch-recompute case¶

The flow that motivates the port: a batch lands, then one set-based statement recomputes over it. The compute is expressed once, in the registered statement — here modelled in-process so the recipe runs without a warehouse:

def _recompute_region_totals(
    params: RecomputeWindow,
    state: MockState,
) -> ExecResult[None]:
    """Model the set-based recompute the production SQL performs: read the ingested batch,
    aggregate per region in one pass, materialize the read model. Returns the rows written.
    """

    sales = state.analytics_ingest_log.get("sales", [])

    totals: dict[str, int] = {}
    for row in sales:
        totals[row["region"]] = totals.get(row["region"], 0) + int(row["amount"])

    state.analytics_query_hits["sales"] = {
        "by_region": [{"region": r, "total": t} for r, t in sorted(totals.items())]
    }

    return ExecResult(affected_count=len(totals))

In production that handler is a single Postgres statement; the calling code does not change. Field encryption applies to a procedure's params when a keyring is wired — the same field-level encryption the document and analytics paths use.