API reference

The API is split into the three buckets described in Module layout — declarative primitives at codegen_database, migration glue at codegen_database.alembic, and pre-built features at codegen_database.ext.

Declarative primitives (codegen_database.*)

codegen_database: configuration-driven PostgreSQL framework.

This top-level module exposes only the declarative primitives you compose to describe a schema (bases, views, columns, foreign keys, indexes, checks, types, plugin/extension base classes, config).

Three-bucket boundary:

• Declarative primitives — right here at codegen_database.

• Migration glue — see codegen_database.alembic (wired into your project’s env.py).

• Pre-built features composed from the primitives — see codegen_database.ext (ledgers, audit, state machines, refresh, cron, row-level security, etc.).

These are the building blocks you compose to describe a schema.

Declarative base classes for codegen_database-managed models and views.

Subclass CodegenDatabaseBase once to create a project-level base, then define model classes by inheriting from it. The codegen_database plugin pipeline runs automatically when each class body is executed.

Subclass CodegenDatabaseView once to create a project-level view base, then define views by inheriting from it. The view is registered for Alembic and the class is ORM-mapped so that select(MyView) works.

class CodegenDatabaseBase

Base class for declarative codegen_database models.

Define a project-level base by subclassing. Metadata is auto-created for your model with codegen_database naming conventions if you don’t provide it:

class Base(CodegenDatabaseBase):
    pass  # Base.metadata created automatically

Override metadata to use custom naming conventions:

class Base(CodegenDatabaseBase):
    metadata = MetaData(naming_convention=...)

Define tables and views in the same inheritance tree. A class with __query__ is treated as a view; without it, the codegen_database plugin pipeline runs and creates a table:

class Locations(Base):
    __tablename__ = "locations"
    __table_args__ = {"schema": "public"}

    name = Column(String, nullable=False)

class LocationStats(Base):
    __tablename__ = "location_stats"
    __table_args__ = {"schema": "public"}
    __query__ = select(
        Locations.name,
        func.count().label("count"),
    ).group_by(Locations.name)

Columns may be declared with either Column(...) or the SQLAlchemy 2.0 mapped_column(...) form (optionally annotated with Mapped[X] for type checkers). mapped_column always requires an explicit SQL type — codegen_database does not run SA’s annotation-driven inference pass.

When the class body executes the appropriate pipeline runs and the class is ORM-mapped so that select(LocationStats) works.

Base-level attributes (set on your Base subclass):

• metadata: Optional. Auto-created with codegen_database naming conventions if not provided.

• codegen_database_config: Optional CodegenDatabaseConfig. If omitted, falls back to metadata.info["codegen_database_config"] so you only need to set the config once.

Per-model attributes resolved through the MRO:

• __factory__: Factory class whose _INTERNAL_PLUGINS are used (default CodegenDatabaseSimple). Set to e.g. CodegenDatabaseAppendOnly for append-only semantics.

• __plugins__: List of plugin instances appended to the resolved plugin list.

class CodegenDatabaseFunctionMixin

Mark a CodegenDatabaseBase subclass as a PostgreSQL function.

Combine with your project base to define functions alongside tables and views:

class InventoryAdjust(CodegenDatabaseFunctionMixin, Base):
    __table_args__ = {"schema": "private"}
    __funcspec__ = ledger_event_function(Inventory, adjust_event)

__funcspec__ must be a CodegenDatabaseFunctionSpec (returned by ledger_event_function(), chart-function builders, etc.). The schema is taken from __table_args__, not from the spec.

Subclasses gain a call() classmethod that wraps the generated SQL function in a select(func.<schema>.<name>(...)) and dispatches via db.execute(...) – so callers never have to spell out the schema, function name, or param order:

await InventoryAdjust.call(
    db,
    warehouse="w1",
    sku="abc",
    value=10,
)

classmethod call(db, **kwargs)

Invoke the generated SQL function.

Builds select(func.<schema>.<name>(*args)) from the declared __funcspec__ and dispatches it via db.execute(...). Kwargs are keyed by the declared param names (without the p_ prefix that the SQL function uses internally); enum.Enum values are coerced to .value so callers can pass typed enums directly.

Parameters:

• db (Any) – Any object with an execute(stmt) method – Session, AsyncSession, or Connection. codegen_database does no awaiting itself; if execute returns an awaitable, await the return value at the call site.

• **kwargs (Any) – Function arguments keyed by param name (without p_ prefix).

Return type:

Any

Returns:

Whatever db.execute(stmt) returns – a Result for sync sessions, an Awaitable[Result] for async ones.

Raises:

KeyError – If a declared param has no matching kwarg.

class CodegenDatabaseView

Base class for declarative codegen_database views.

Define a project-level view base by subclassing and setting metadata to the same instance used by your model base. Views and tables must share metadata so that Alembic sees everything in one place:

class ViewBase(CodegenDatabaseView):
    metadata = Base.metadata

Then define view classes by inheriting from that base:

# Columns inferred from the query (names and best-effort types).
class CustomerStats(ViewBase):
    __tablename__ = "customer_order_stats"
    __table_args__ = {"schema": "public"}
    __query__ = (
        select(
            orders.c.customer_id,
            func.count().label("order_count"),
            func.sum(orders.c.total).label("order_total"),
        )
        .group_by(orders.c.customer_id)
    )

# Columns declared explicitly (accurate types, self-documenting).
class InvoiceStats(ViewBase):
    __tablename__ = "customer_invoice_stats"
    __table_args__ = {"schema": "public"}

    customer_id = Column(Integer, nullable=False)
    invoice_count = Column(Integer)
    invoiced_total = Column(Numeric(10, 2))

    __query__ = (
        select(
            invoices.c.customer_id,
            func.count().label("invoice_count"),
            func.sum(invoices.c.amount).label("invoiced_total"),
        )
        .group_by(invoices.c.customer_id)
    )

# Materialized view.
class ProductSummary(ViewBase):
    __tablename__ = "product_summary"
    __table_args__ = {"schema": "public"}
    __options__ = ViewOptions(materialized=True)
    __query__ = select(...)

When the class body executes the view DDL is registered on the metadata for Alembic autogeneration, and the class is ORM-mapped so that select(CustomerStats) works.

Column resolution:

If the class declares Column attributes they are used as-is — this is the preferred form because it gives accurate types for the ORM mapper and is self-documenting.

If no columns are declared the list is derived from __query__.selected_columns. Types are inherited from the source columns where possible (e.g. direct column references), but aggregates and expressions that lose type information are mapped as NullType. A surrogate primary key is chosen automatically: the column named id if present, otherwise the first column.

Per-view ``__options__`` (own class dict only):

Set __options__ = ViewOptions(materialized=True) to create a PostgreSQL materialized view.

class CodegenDatabaseViewMixin

Mixin that marks a CodegenDatabaseBase subclass as a SQL view.

Combine with your project base to define views in the same class hierarchy as tables:

class OrderStats(CodegenDatabaseViewMixin, Base):
    __tablename__ = "order_stats"
    __table_args__ = {"schema": "public"}
    __query__ = select(Orders.customer_id, func.count().label("n"))

class ViewOptions(materialized=False)

Options for declarative views.

Set on a CodegenDatabaseViewMixin or CodegenDatabaseView subclass via __options__.

materialized

If True, creates a PostgreSQL materialized view instead of a regular view.

map_model(cls, parent_cls=None)

Run the codegen_database plugin pipeline for cls and ORM-map it.

Called from CodegenDatabaseBase.__init_subclass__() once per model class (those that have a __tablename__), and ad hoc for classes declared outside that hierarchy.

Parameters:

• cls (type) – The model class being defined.

• parent_cls (type | None) – an optional parent class to merge with the input cls. can be helpful when the input class cannot inherit from the parent class directly. This is particularly helpful for places where the “child class” model is defined in one library and would otherwise have to subclass the codegen database class at runtime, which would cause issues when trying to use that class in a query, since the class would not be equivalent to the sqla-ified class.

Raises:

CodegenDatabaseValidationError – On missing schema or plugin misconfiguration.

Return type:

None

Global codegen_database configuration.

class CodegenDatabaseConfig(plugins=<factory>, extensions=<factory>, auto_discover=True, utility_schema='codegen_database')

Global plugin and extension registry.

Plugins registered here are prepended to every factory’s resolved plugin list, so they run before factory-specific plugins.

Extensions bundle plugins, metadata hooks, Alembic hooks, and CLI commands into a single unit.

Example:

config = CodegenDatabaseConfig()
config.register(TimestampPlugin(), TenantPlugin())

CodegenDatabaseSimple(
    "users", "public", metadata, ..., config=config
)

Parameters:

• plugins (list[Plugin]) – Global plugins prepended to every factory.

• extensions (list[CodegenDatabaseExtension]) – Manually registered extension instances.

• auto_discover (bool) – Whether to discover extensions via entry points. Defaults to True.

• utility_schema (str) – PostgreSQL schema for codegen_database-managed utility functions (e.g. ledger_apply_state). Defaults to "codegen_database". Override only if your project already uses a schema named "codegen_database".

property all_plugins: list[Plugin]

Return extension plugins + direct plugins.

Extension plugins are prepended before direct plugins.

Returns:

Combined plugin list.

register(*plugins)

Register one or more plugins globally.

Parameters:

*plugins (Plugin) – Plugin instances to add.

Return type:

CodegenDatabaseConfig

Returns:

self for chaining.

use(*extensions)

Register one or more extensions.

Parameters:

*extensions (CodegenDatabaseExtension) – Extension instances to add.

Return type:

CodegenDatabaseConfig

Returns:

self for chaining.

DEFAULT_UTILITY_SCHEMA = 'codegen_database'

Default schema for codegen_database-managed utility objects – the CodegenDatabaseConfig.utility_schema default and the fallback resolve_utility_schema() uses when no config is present.

resolve_utility_schema(metadata)

Return the codegen_database utility schema for metadata.

Reads utility_schema off the CodegenDatabaseConfig registered at metadata.info["codegen_database_config"], falling back to DEFAULT_UTILITY_SCHEMA when no config is present.

Any codegen_database-managed object that needs a shared home – utility functions, the autogenerated-identifier pointer table – routes through here so a single utility_schema setting governs them all.

Parameters:

metadata (MetaData) – SQLAlchemy MetaData to inspect.

Return type:

str

Returns:

The schema name for codegen_database-managed utility objects.

Check constraint support for codegen_database dimensions.

Provides CodegenDatabaseCheck, a declarative check constraint that uses {column_name} markers in its expression. Plugins resolve these markers to the appropriate column references depending on the dimension type (table-level for simple/append-only, NEW.col for EAV trigger-based enforcement).

class CodegenDatabaseCheck(expression, name)

A declarative check constraint with {col} markers.

Parameters:

• expression (str) – Constraint expression using {column_name} markers, e.g. "{price} > 0".

• name (str) – Required constraint name — no auto-naming.

column_names()

Extract {name} markers from the expression.

Return type:

list[str]

Returns:

List of column names referenced in the expression, in order of first appearance with duplicates removed.

resolve(mapping)

Replace each {col} with mapping(col).

Parameters:

mapping (Callable[[str], str]) – A callable that maps column names to their resolved form (e.g. identity for table-level, lambda c: f"NEW.{c}" for triggers).

Return type:

str

Returns:

The resolved SQL expression.

collect_checks(schema_items)

Filter CodegenDatabaseCheck instances from a schema items list.

Parameters:

schema_items (list) – Mixed list of Column, CodegenDatabaseCheck, and other schema items.

Return type:

list[CodegenDatabaseCheck]

Returns:

Only the CodegenDatabaseCheck items, in their original order.

Index support for codegen_database dimensions.

Provides CodegenDatabaseIndex, a declarative index definition that mirrors sqlalchemy.Index and uses {column_name} markers for column references.

class CodegenDatabaseIndex(name, *expressions, unique=False, **kw)

A declarative index definition with {col} markers.

Mirrors the sqlalchemy.Index constructor signature:

CodegenDatabaseIndex("idx_name", "{col1}", "{col2}",
             unique=True, postgresql_using="btree")

Simple column references ("{name}") and functional expressions ("lower({name})") are both supported. Extra keyword arguments are passed through to the underlying sqlalchemy.Index.

Parameters:

• name (str) – Required index name.

• *expressions (str) – Index expressions using {column_name} markers.

• unique (bool) – Whether to create a unique index.

• **kw (Any) – Passed through to sqlalchemy.Index (e.g. postgresql_using, postgresql_where).

column_names()

Extract {name} markers from all expressions.

Return type:

list[str]

Returns:

Column names in order of first appearance, deduplicated across all expressions.

resolve(mapping)

Replace {col} markers in each expression.

Parameters:

mapping (Callable[[str], str]) – A callable that maps column names to their resolved form.

Return type:

list[str]

Returns:

List of resolved expression strings.

collect_indices(schema_items)

Filter CodegenDatabaseIndex instances from schema items.

Parameters:

schema_items (list) – Mixed list of schema items.

Return type:

list[CodegenDatabaseIndex]

Returns:

Only the CodegenDatabaseIndex items, in original order.

trigram_indexes(*columns, table=None, method='gin')

Build one pg_trgm index per column for fuzzy text search.

A pg_trgm operator-class index is what makes the % similarity operator and ILIKE substring matches fast – without one, a trigram search falls back to a sequential scan. Pass the text columns a resource searches and splat the result into a dimension’s schema_items; the TableIndexPlugin materializes each into a real index:

schema_items=[
    Column("name", String),
    Column("sku", String),
    *trigram_indexes("name", "sku", table="product"),
]

Every index is USING <method> (<col> <method>_trgm_ops) – the operator class is what binds the index to the trigram operators, so a plain USING gin (col) would not serve % / ILIKE lookups.

The pg_trgm extension is registered by default – configure_metadata adds it via codegen_database.pg_extension.register_default_pg_extensions(), so the next autogenerated migration emits the CREATE EXTENSION when the database lacks it.

Parameters:

• *columns (str) – Text column names to index for trigram search.

• table (str | None) – Optional table name woven into each index name so it stays unique schema-wide – index names are database-global. Omitted yields ix__<col>__trgm; given, ix__<table>__<col>__trgm.

• method (str) – Index access method – "gin" (default, the usual choice) or "gist". The matching <method>_trgm_ops operator class is applied.

Return type:

list[CodegenDatabaseIndex]

Returns:

One CodegenDatabaseIndex per column, in input order.

Raises:

CodegenDatabaseValidationError – If columns is empty or contains a duplicate, or method is neither "gin" nor "gist".

Foreign key support for codegen_database dimensions.

Provides CodegenDatabaseForeignKey — an inline single-column FK passed directly to a Column(...) constructor, analogous to SQLAlchemy’s ForeignKey. Accepts either a two-part "dimension.column" reference (resolved via the registry) or a fully-qualified "schema.table.column" reference.

class CodegenDatabaseForeignKey(reference, *, ondelete=None, onupdate=None)

Inline single-column FK with codegen_database dimension resolution.

Pass directly to a Column constructor like SQLAlchemy’s ForeignKey. The reference string accepts two formats:

• "dimension.column" — resolved via the dimension registry at factory time (two dot-separated parts):

  Column("user_id", Integer, CodegenDatabaseForeignKey("users.id"))
  

• "schema.table.column" — passed through directly:

  Column("user_id", Integer,
         CodegenDatabaseForeignKey("public.users_raw.id"))
  

Parameters:

• reference (str) – Target reference string.

• ondelete (str | None) – ON DELETE action (e.g. "CASCADE").

• onupdate (str | None) – ON UPDATE action (e.g. "CASCADE").

class DimensionRef(schema, table)

Registry entry for a dimension’s FK-targetable table.

Stored in metadata.info["codegen_database_dimensions"] keyed by dimension name (tablename).

Parameters:

• schema (str) – PostgreSQL schema name.

• table (str) – Physical table name for FK targets.

defer_fk_resolution(metadata, dimension, table, column_name, fk)

Park an FK declaration until dimension is registered.

Factories run at class-creation time, so a two-part reference can name a dimension whose model simply hasn’t been imported yet. register_dimension() materializes parked entries the moment the dimension arrives; validate_fks_resolved() reports any that never do.

Parameters:

• metadata (MetaData) – SQLAlchemy MetaData instance.

• dimension (str) – The (not yet registered) dimension name.

• table (Table) – The table the constraint will be appended to.

• column_name (str) – The referencing column on table.

• fk (CodegenDatabaseForeignKey) – The original inline FK marker.

Return type:

None

materialize_fk(metadata, table, column_name, fk, resolved_ref)

Append the ForeignKeyConstraint for an inline FK marker.

Parameters:

• metadata (MetaData) – SQLAlchemy MetaData instance (for target validation).

• table (Table) – The table to append the constraint to.

• column_name (str) – The referencing column on table.

• fk (CodegenDatabaseForeignKey) – The inline FK marker carrying referential actions.

• resolved_ref (str) – Fully qualified "schema.table.column" target.

Return type:

None

register_dimension(metadata, name, ref)

Register a dimension for FK resolution.

Also materializes any FK declarations that were deferred because they referenced name before it was registered – model import order doesn’t constrain who may reference whom.

Parameters:

• metadata (MetaData) – SQLAlchemy MetaData instance.

• name (str) – Dimension name (tablename).

• ref (DimensionRef) – The dimension’s FK target info.

Return type:

None

resolve_fk_reference(metadata, reference)

Resolve a "dimension.column" reference.

Looks up the dimension name in the registry and expands it to "schema.table.column".

Parameters:

• metadata (MetaData) – SQLAlchemy MetaData for registry lookup.

• reference (str) – Two-part "dimension.column" string.

Return type:

str

Returns:

Fully qualified "schema.table.column" string.

Raises:

CodegenDatabaseValidationError – If the reference does not contain exactly one dot, or names an unknown dimension.

validate_fk_target(resolved_ref, metadata)

Validate a fully-qualified FK target against the metadata.

Only validates when the target table is present in metadata (i.e. it is a codegen_database-managed table). External tables are silently skipped.

Parameters:

• resolved_ref (str) – "schema.table.column" string.

• metadata (MetaData) – SQLAlchemy MetaData instance.

Raises:

CodegenDatabaseValidationError – If the target column does not exist on a known table.

Return type:

None

validate_fks_resolved(metadata)

Fail if any deferred FK never found its dimension.

Call after every model module is imported (the alembic configure_metadata hook does) – a leftover entry means the reference names a dimension that doesn’t exist, and silently omitting the constraint would be far worse than failing here.

Parameters:

metadata (MetaData) – SQLAlchemy MetaData instance.

Raises:

CodegenDatabaseValidationError – A two-part FK reference names a dimension that was never registered.

Return type:

None

Declarative and imperative PostgreSQL function registration.

CodegenDatabaseFunction works both imperatively and as a declarative base class, mirroring how CodegenDatabaseView and CodegenDatabasePlainView work for views.

class CodegenDatabaseFunction(name, schema, definition, *, metadata=None, returns='void', language='sql', parameters=None, security=FunctionSecurity.invoker, volatility=FunctionVolatility.VOLATILE)

Register a PostgreSQL function on SQLAlchemy metadata.

Works both imperatively (direct instantiation) and as a declarative base class (subclassing).

Imperative usage (metadata read from class when omitted):

class FunctionBase(CodegenDatabaseFunction):
    metadata = Base.metadata

spec = ledger_event_function(source, event)
FunctionBase(
    spec["name"],
    source.ctx.schemaname,
    spec["definition"],
    parameters=spec["parameters"],
    returns=spec["returns"],
    security=spec["security"],
)

Or with explicit metadata:

CodegenDatabaseFunction(
    name="inventory_adjust",
    schema="private",
    definition="...",
    metadata=metadata,
    returns="SETOF private.inventory_raw",
    parameters=[...],
    security=FunctionSecurity.definer,
)

Declarative usage:

class FunctionBase(CodegenDatabaseFunction):
    metadata = metadata

class InventoryAdjust(FunctionBase):
    __funcname__ = "inventory_adjust"
    __table_args__ = {"schema": "private"}
    __definition__ = "..."
    __options__ = FunctionOptions(
        returns="SETOF private.inventory_raw",
        parameters=[...],
        security=FunctionSecurity.definer,
    )

In both cases a sqlalchemy_declarative_extensions.dialects.postgresql.Function is registered on the provided MetaData instance so that Alembic autogeneration picks it up.

Parameters:

• name (str) – Unqualified function name.

• schema (str) – PostgreSQL schema.

• metadata (MetaData | None) – SQLAlchemy MetaData to register on.

• definition (str) – SQL (or PL/pgSQL) function body.

• returns (str) – Return type string (default "void").

• language (str) – Function language (default "sql").

• parameters (list[FunctionParam] | None) – List of FunctionParam instances from sqlalchemy_declarative_extensions.dialects.postgresql (default []).

• security (FunctionSecurity) – Security mode — FunctionSecurity.invoker or FunctionSecurity.definer (default FunctionSecurity.invoker).

• volatility (FunctionVolatility) – Volatility classification (default FunctionVolatility.VOLATILE).

class CodegenDatabaseFunctionSpec

Specification dict for a PostgreSQL function.

Returned by spec-builders such as ledger_event_function(), construct_ledger_chart_function(), and construct_date_bin_function(). Pass directly as __funcspec__ on a CodegenDatabaseFunctionMixin subclass, or unpack into CodegenDatabaseFunction for imperative registration:

class InventoryAdjust(CodegenDatabaseFunctionMixin, Base):
    __table_args__ = {"schema": "private"}
    __funcspec__ = ledger_event_function(inventory, adjust_event)

name

Unqualified function name.

definition

SQL function body.

language

Function language ("sql", "plpgsql", …).

parameters

List of FunctionParam instances.

returns

Return type string (e.g. "SETOF schema.table").

security

Security mode.

volatility

Optional volatility (STABLE / IMMUTABLE / VOLATILE). Defaults to VOLATILE when omitted.

Note

Schema is intentionally excluded. Pass it via __table_args__ (declarative) or read it from the source’s ctx.schemaname (imperative).

class FunctionOptions(returns='void', language='sql', parameters=<factory>, security=FunctionSecurity.invoker, volatility=FunctionVolatility.VOLATILE)

Options for CodegenDatabaseFunction declarative subclasses.

Set on a subclass via __options__.

returns

Return type string.

language

Function language ("sql", "plpgsql", …).

parameters

List of FunctionParam instances from sqlalchemy_declarative_extensions.dialects.postgresql.

security

FunctionSecurity.invoker or FunctionSecurity.definer.

volatility

Volatility classification.

Custom SQLAlchemy column types for codegen_database.

• TextEnum / IntEnum – persist a Python enum as TEXT / INTEGER.

• EncryptedText – a Fernet-encrypted-at-rest TEXT column (requires the encrypted extra for cryptography).

• COORDINATE / Coordinate – a geographic latitude / longitude value stored as a PostGIS geography(Point,4326), round-tripping as a Coordinate frozen dataclass.

• VALUE_TYPES / build_value_type() – the integer / numeric / decimal value-column vocabulary the ledger factories type their amount columns from, with precision / scale pass-through to NUMERIC.

• STDADDR / StdAddr – the PostGIS stdaddr composite type (from the address_standardizer extension), round-tripping as a StdAddr frozen dataclass of str parts ("" = NULL sub-field).

All names are re-exported here so callers keep importing from codegen_database.types regardless of which submodule owns them.

class COORDINATE(*args, **kwargs)

A geographic coordinate stored as a PostGIS geography point.

The underlying column is geography(Point,4326) (WGS84). Only Coordinate is accepted on input.

Coordinates are written with ST_GeogFromText and read with ST_AsText.

Requires the postgis extension: register PostgisExtension on your config so Alembic creates it.

Example:

from sqlalchemy import Column
from codegen_database.types import COORDINATE, Coordinate

Column("coordinate", COORDINATE(), nullable=False)

bind_expression(bindvalue)

Wrap the bound WKT in ST_GeogFromText on the way in.

Return type:

Any

column_expression(column)

Read the column back as WKT via ST_AsText.

Return type:

Any

process_bind_param(value, dialect)

Serialize a Coordinate to POINT(longitude latitude) WKT.

Return type:

str | None

process_literal_param(value, dialect)

Render as an inline SQL literal (for literal_binds).

Returns the quoted WKT (well known text) bind_expression() then wraps it in ST_GeogFromText(...).

Return type:

str

process_result_value(value, dialect)

Parse POINT(longitude latitude) WKT into a Coordinate.

Return type:

Coordinate | None

class Coordinate(*, latitude, longitude)

A geographic coordinate: latitude / longitude in degrees.

Values are Decimal for exact handling in Python. This is a Python-side representation only: PostGIS stores the point as float8 (double precision), so a value read back is a Decimal of the rounded double, not necessarily the exact value originally written.

class EncryptedText(key)

Store a string Fernet-encrypted in a TEXT column.

Parameters:

key (str | Callable[[], str]) – The encryption key material, or a zero-argument callable returning it. A callable is invoked on every encrypt / decrypt, so keys read from the environment resolve lazily – pass e.g. lambda: os.environ["TOKEN_KEY"].

Example:

import os
from sqlalchemy.orm import Mapped, mapped_column
from codegen_database.types import EncryptedText

class Connection(Base):
    ...
    access_token: Mapped[str] = mapped_column(
        EncryptedText(key=lambda: os.environ["TOKEN_KEY"]),
    )

Reading a value encrypted under a different key raises cryptography.fernet.InvalidToken – rotating the key makes previously stored values undecryptable, so treat the key like a database credential.

process_bind_param(value, dialect)

Encrypt value on its way into the database.

Return type:

str | None

process_result_value(value, dialect)

Decrypt a stored ciphertext back to plaintext.

Return type:

str | None

class IntEnum(enum_class)

Store a Python enum as an integer in PostgreSQL.

Values are persisted as the enum member’s .value (which must be an integer) and coerced back to the corresponding Python enum member on load. The underlying column is INTEGER.

Parameters:

enum_class (type[Enum]) – The enum.Enum subclass to coerce values to and from. All member values must be integers.

Example:

import enum
from sqlalchemy import Column
from codegen_database.types import IntEnum

class Priority(enum.Enum):
    LOW = 1
    MEDIUM = 2
    HIGH = 3

Column("priority", IntEnum(Priority), nullable=False)

process_bind_param(value, dialect)

Convert a Python enum member to its integer value.

Return type:

int | None

process_result_value(value, dialect)

Convert a stored integer back to a Python enum.

Return type:

Enum | None

class STDADDR(*args, **kwargs)

A PostGIS stdaddr composite column.

The underlying column is the native stdaddr type from the address_standardizer extension. Values round-trip as StdAddr frozen dataclasses; only StdAddr (or None) is accepted on input. "" parts are stored as NULL sub-fields and come back as "".

Conversion is delegated to PostgreSQL via JSON, so there is no hand-written composite (de)serialization:

• On read, column_expression() wraps the column in to_jsonb(...); the driver returns a dict that process_result_value() splats into a StdAddr.

• On write, process_bind_param() emits a JSON object and bind_expression() feeds it through jsonb_populate_record(NULL::stdaddr, ...).

Requires the extension to be installed:

CREATE EXTENSION IF NOT EXISTS address_standardizer

Example:

from sqlalchemy import Column
from codegen_database.types import STDADDR, StdAddr

Column("address", STDADDR(), nullable=True)
# store StdAddr(house_num="123", name="MAIN", suftype="ST")

bind_expression(bindvalue)

Build the stdaddr composite from the bound JSON object.

Return type:

Any

column_expression(colexpr)

Read the column as jsonb so the driver returns a mapping.

type_=self keeps the wrapped expression typed as STDADDR so process_result_value() still runs on the decoded value (the driver decodes jsonb to a dict by its server OID regardless of the SQLAlchemy type).

Return type:

Any

process_bind_param(value, dialect)

Serialize a StdAddr to a JSON object.

"" parts become JSON null (NULL sub-fields). Accepts any object so a clear error is raised for callers who pass something other than a StdAddr.

Return type:

str | None

process_result_value(value, dialect)

Build a StdAddr from the to_jsonb mapping.

NULL sub-fields become "".

Return type:

StdAddr | None

class StdAddr(*, building='', house_num='', predir='', qual='', pretype='', name='', suftype='', sufdir='', ruralroute='', extra='', city='', state='', country='', postcode='', box='', unit='')

A parsed PostGIS stdaddr value.

Each attribute maps to one column of the PostgreSQL stdaddr composite type, in declaration order, and is a str where "" means the sub-field is absent (stored as NULL). All fields default to "".

The field set and order mirror the composite type defined by address_standardizer:

CREATE TYPE stdaddr AS (
    building text, house_num text, predir text, qual text,
    pretype text, name text, suftype text, sufdir text,
    ruralroute text, extra text, city text, state text,
    country text, postcode text, box text, unit text
);

class TextEnum(enum_class)

Store a Python enum as plain text in PostgreSQL.

Values are persisted as the enum member’s .value (which must be a string) and coerced back to the corresponding Python enum member on load. No PostgreSQL ENUM type is created — the underlying column is TEXT.

Parameters:

enum_class (type[Enum]) – The enum.Enum subclass to coerce values to and from.

Example:

import enum
from sqlalchemy import Column
from codegen_database.types import TextEnum

class Color(enum.Enum):
    RED = "red"
    GREEN = "green"
    BLUE = "blue"

Column("color", TextEnum(Color), nullable=False)

process_bind_param(value, dialect)

Convert a Python enum member to its string value.

Return type:

str | None

process_result_value(value, dialect)

Convert a stored text value back to a Python enum.

Return type:

Enum | None

build_value_type(value_type, *, precision=None, scale=None)

Instantiate the SQLAlchemy type for a named value column.

Parameters:

• value_type (Literal['integer', 'numeric', 'decimal']) – One of ValueType.

• precision (int | None) – Total number of digits for a NUMERIC column (the NUMERIC(precision, scale) first argument). Only valid for the "numeric" / "decimal" types.

• scale (int | None) – Number of digits after the decimal point. Requires precision – NUMERIC cannot fix a scale without a precision.

Return type:

TypeEngine

Returns:

A configured TypeEngine instance ready to drop into a Column.

Raises:

CodegenDatabaseValidationError – If value_type is unknown, if precision/scale are supplied for a non-NUMERIC type, or if scale is given without precision.

Generic view factories.

class CodegenDatabaseMaterializedView(name, schema, metadata, query)

Create a materialized view with an auto-generated refresh.

After construction, self.table is a joinable SQLAlchemy Table whose columns mirror the query.

Parameters:

• name (str) – View name.

• schema (str) – PostgreSQL schema for the view.

• metadata (MetaData) – SQLAlchemy MetaData to register on.

• query (Select) – A SQLAlchemy Select defining the view body.

class CodegenDatabasePlainView(name, schema, metadata, query)

Create a plain PostgreSQL view from a SQLAlchemy select.

Prefer the declarative CodegenDatabaseView base class for new views; this class exists for cases where an imperative registration with a self.table handle is needed.

After construction, self.table is a joinable SQLAlchemy Table whose columns mirror the query.

Parameters:

• name (str) – View name.

• schema (str) – PostgreSQL schema for the view.

• metadata (MetaData) – SQLAlchemy MetaData to register on.

• query (Select) – A SQLAlchemy Select defining the view body.

Plugin base class and helpers for codegen_database factory extensions.

class Dynamic(attr)

Reference to an instance attribute that holds the actual ctx key.

Use inside produces() / requires() decorators when the ctx key name is determined at construction time rather than being a fixed string:

@produces(Dynamic("table_key"))
@requires("raw")
class MyPlugin(Plugin):
    def __init__(self, table_key: str = "result") -> None:
        self.table_key = table_key

    def run(self, ctx: FactoryContext) -> None:
        ctx[self.table_key] = build(ctx["raw"])

The factory resolves each Dynamic via getattr(instance, attr) when building the dependency graph.

class MinPGVersion(version)

Minimum PostgreSQL version requirement for a plugin.

Use inside requires() to declare that a plugin needs a specific PostgreSQL major version:

@requires(MinPGVersion(18))
@produces("pk_columns")
class UUIDV7PKPlugin(Plugin):
    ...

The factory stores the requirement as min_pg_version on the class. Call check_pg_version() with the connected server’s major version to validate before applying DDL.

class Plugin

Base class for codegen_database factory plugins.

Each plugin implements run to perform its work. Execution order is determined by topological sort using the produces() and requires() class decorators.

Declaring dependencies:

@produces(Dynamic("out_key"))
@requires("primary")
class MyPlugin(Plugin):
    def __init__(self, out_key: str = "result") -> None:
        self.out_key = out_key

    def run(self, ctx: FactoryContext) -> None:
        ctx[self.out_key] = transform(ctx["primary"])

Plugins communicate through ctx using string keys. Use the singleton() decorator to declare that at most one plugin of a given group may appear in any resolved plugin list.

resolved_produces()

Return the ctx keys this plugin writes, with Dynamic refs resolved.

Reads the _produces list set by the produces() decorator and substitutes each Dynamic with getattr(self, attr).

Return type:

list[str]

Returns:

List of ctx key strings this plugin will write to.

resolved_requires()

Return the ctx keys this plugin reads, with Dynamic refs resolved.

Reads the _requires list set by the requires() decorator and substitutes each Dynamic with getattr(self, attr).

Return type:

list[str]

Returns:

List of ctx key strings this plugin expects to already be set.

run(ctx)

Execute this plugin’s work against ctx.

The factory calls this once per plugin, after topological sorting by produces() / requires().

Parameters:

ctx (FactoryContext) – The factory context.

Return type:

None

class PluginCollection

Generator for multiple plugins

This is often useful if you want to define a plugin that needs to run at two different points in the dependency graph. For example, a plugin that needs to modify both the view and the backing table might need to run before and after the index plugins.

check_pg_version(server_version, plugins)

Raise if any plugin requires a newer PostgreSQL version.

Call this with the server’s major version (e.g. conn.dialect.server_version_info[0]) to get an early, clear error instead of a cryptic “function does not exist” from PostgreSQL.

Parameters:

• server_version (int) – Major version of the connected server.

• plugins (list[Plugin]) – The resolved plugin list to check.

Raises:

CodegenDatabaseValidationError – When a plugin’s min_pg_version exceeds server_version.

Return type:

None

produces(*keys)

Declare the ctx keys this plugin’s run method writes.

Applied as a class decorator, alongside requires() and singleton():

@produces(Dynamic("table_key"))
class MyTablePlugin(Plugin):
    ...

Parameters:

*keys (str | Dynamic) – Ctx key strings or Dynamic references to instance attributes that hold the actual key names.

Return type:

Callable[[TypeVar(T, bound= type[Plugin])], TypeVar(T, bound= type[Plugin])]

Returns:

A class decorator that attaches _produces to the class.

Raises:

TypeError – If a Dynamic attr name is not an __init__ parameter.

requires(*keys)

Declare the ctx keys this plugin’s run method reads.

Applied as a class decorator, alongside produces() and singleton(). Accepts MinPGVersion sentinels to declare a minimum PostgreSQL version requirement:

@requires(MinPGVersion(18), "pk_columns")
class MyPlugin(Plugin):
    ...

Parameters:

*keys (str | Dynamic | MinPGVersion) – Ctx key strings, Dynamic references, or MinPGVersion version requirements.

Return type:

Callable[[TypeVar(T, bound= type[Plugin])], TypeVar(T, bound= type[Plugin])]

Returns:

A class decorator that attaches _requires to the class and sets min_pg_version if any MinPGVersion sentinel is present.

Raises:

TypeError – If a Dynamic attr name is not an __init__ parameter.

singleton(group)

Declare that at most one plugin of group may appear.

The factory raises CodegenDatabaseValidationError at construction time if two plugins with the same group name are present in the resolved plugin list.

Example:

@singleton("__pk__")
class MyPKPlugin(Plugin):
    ...

Parameters:

group (str) – Arbitrary group identifier. By convention, built-in groups use dunder names ("__pk__", "__table__").

Return type:

Callable[[TypeVar(T, bound= type[Plugin])], TypeVar(T, bound= type[Plugin])]

Returns:

A class decorator that sets singleton_group on the class and registers the singleton validator.

Extension base class and discovery for codegen_database.

class CodegenDatabaseExtension(name)

Base class for codegen_database extensions.

An extension bundles plugins, metadata hooks, Alembic hooks, and CLI commands into a single installable unit.

Subclasses override hook methods to participate in the codegen_database lifecycle. Extensions declare inter-extension dependencies via the depends_on class variable.

Example:

@dataclass
class MyExtension(CodegenDatabaseExtension):
    name: str = "my-ext"

    def plugins(self) -> list[Plugin]:
        return [MyGlobalPlugin()]

    def configure_metadata(self, metadata: MetaData) -> None:
        # register roles, grants, schemas, etc.
        ...

configure_alembic()

Register custom Alembic renderers or rewriters.

Override to hook into Alembic setup. Called by alembic_hook.

Return type:

None

configure_metadata(metadata)

Configure metadata-level objects.

Override to register roles, grants, schemas, or other metadata-level objects. Called by configure_metadata.

Parameters:

metadata (MetaData) – The SQLAlchemy MetaData being configured.

Return type:

None

plugins()

Global plugins prepended to every factory.

Return type:

list[Plugin]

Returns:

List of plugin instances. Empty by default.

register_cli(app)

Add subcommands during CLI setup.

Parameters:

app (object) – The typer.Typer application instance.

Return type:

None

validate(registered_names)

Validate after all extensions are loaded.

Override to check that required peer extensions are present or that configuration is consistent.

Parameters:

registered_names (frozenset[str]) – Names of all loaded extensions.

Return type:

None

discover_extensions()

Discover extensions via the codegen_database.ext entry point group.

Return type:

dict[str, type[CodegenDatabaseExtension]]

Returns:

Mapping of extension name to extension class.

validate_extension_deps(extensions)

Check that every extension’s depends_on is satisfied.

Parameters:

extensions (list[CodegenDatabaseExtension]) – The resolved list of extension instances.

Raises:

CodegenDatabaseValidationError – If a dependency is missing.

Return type:

None

Declarative PostgreSQL extension management for codegen_database.

PGExtension describes a PostgreSQL extension that the application requires. register_pg_extension() stores it in metadata.info so the Alembic comparator can emit CREATE EXTENSION IF NOT EXISTS when the extension is missing from the database.

Extensions are never dropped automatically. Removing an extension from metadata simply stops codegen_database from ensuring it is present; the extension itself remains installed until a DBA drops it manually.

DEFAULT_PG_EXTENSIONS (currently pg_trgm) are registered on every metadata by codegen_database.alembic.register.configure_metadata() – no opt-in needed; a manual registration of the same name wins.

Usage:

from codegen_database.pg_extension import PGExtension, register_pg_extension

# Standalone registration
register_pg_extension(metadata, PGExtension("btree_gist"))
register_pg_extension(metadata, PGExtension("pg_cron", schema="cron"))

# Alembic wiring (call once in env.py)
from codegen_database.pg_extension import (
    register_pg_extension_alembic_events,
)
register_pg_extension_alembic_events()

Alembic autogenerate then emits:

op.execute("CREATE EXTENSION IF NOT EXISTS btree_gist")

class CreateExtensionOp(extension)

Alembic operation: create a PostgreSQL extension.

Inherits from alembic.operations.ops.MigrateOperation so that codegen_database’s Alembic rewriter passes it through during process_revision_directives traversal without error.

reverse()

Return a no-op downgrade: extensions are never dropped automatically.

The downgrade renderer emits a SQL comment instead of a DROP statement so that autogenerate does not accidentally remove an extension that other objects may depend on.

Return type:

CreateExtensionOp

to_sql()

Return the DDL statement for this operation.

Return type:

list[str]

DEFAULT_PG_EXTENSIONS: tuple[PGExtension, ...] = (PGExtension(name='pg_trgm', schema=None, cascade=False),)

Extensions every codegen_database project gets without opting in. pg_trgm backs the % similarity operator that trigram text search (the default list-endpoint search strategy upstack) and trigram_indexes rely on; it ships with contrib and is harmless when unused, so registering it by default removes a foot-gun – a project that declares searchable text columns no longer 500s at first search because nobody remembered CREATE EXTENSION.

ExtensionOwnedRelations

Relations owned by an installed extension, keyed by (schema, name). schema is always a concrete name (never None); public for the default schema.

alias of set[tuple[str, str]]

class PGExtension(name, schema=None, cascade=False)

Describe a PostgreSQL extension to install.

Parameters:

• name (str) – Extension name, e.g. "btree_gist".

• schema (str | None) – Schema in which to install the extension. When None the database default (usually public) is used.

• cascade (bool) – When True, adds CASCADE so dependent extensions are installed automatically.

to_sql_create()

Render CREATE EXTENSION IF NOT EXISTS DDL.

Return type:

str

Returns:

A complete CREATE EXTENSION SQL string.

class PGExtensions(extensions=<factory>)

Container for all extensions registered on a MetaData.

Stored under metadata.info["pg_extensions"] by register_pg_extension().

classmethod extract(metadata)

Return registered extensions or None if none exist.

Parameters:

metadata (object) – SQLAlchemy MetaData, or None.

Return type:

PGExtensions | None

Returns:

The PGExtensions holder or None.

assert_pg_extension_declared(metadata, name)

Raise if name has not been declared as a required extension.

Call this inside a plugin’s run() method to give a clear error when the user forgot to register the extension in metadata (e.g. via a PGCronExtension).

In the declarative model flow, extensions are configured via configure_metadata which runs after all models are imported. If a plugin calls this check at class-definition time and the extension is not yet in metadata but a CodegenDatabaseConfig is present on metadata.info, the config’s extension hooks are run eagerly so that the check can succeed. The full configure_metadata call in env.py will overwrite any interim state with the final correct values.

Parameters:

• metadata (MetaData) – The MetaData to check.

• name (str) – The PostgreSQL extension name to require.

Raises:

CodegenDatabaseValidationError – If name is not declared and cannot be resolved from the registered config.

Return type:

None

fetch_extension_owned_relations(connection)

Return relations created and owned by installed extensions.

A PostgreSQL extension may create tables, views, sequences, and other relations that it owns (recorded in pg_depend with deptype = 'e'). The canonical example is PostGIS’s spatial_ref_sys table. These relations are part of the extension, not the application schema, so Alembic autogenerate must not emit DROP TABLE / CREATE TABLE for them – dropping one fails outright (cannot drop table spatial_ref_sys because extension postgis requires it).

Parameters:

connection (Connection) – Active database connection.

Return type:

set[tuple[str, str]]

Returns:

A set of (schema, name) pairs for every relation owned by an installed extension. Empty when no extensions own relations.

register_default_pg_extensions(metadata)

Register DEFAULT_PG_EXTENSIONS on metadata.

Called by codegen_database.alembic.register.configure_metadata() so every project wired through the standard env.py hooks gets the defaults; safe to call repeatedly – an extension already declared (by any path) is not added twice.

Parameters:

metadata (MetaData) – SQLAlchemy MetaData to register on.

Return type:

None

register_pg_extension(metadata, ext)

Store ext in metadata for Alembic autogenerate.

Parameters:

• metadata (MetaData) – SQLAlchemy MetaData to register on.

• ext (PGExtension) – The extension to register.

Return type:

None

register_pg_extension_alembic_events()

Register the extension comparator and renderer with Alembic.

Call once in env.py alongside alembic_hook():

from codegen_database.pg_extension import (
    register_pg_extension_alembic_events,
)
register_pg_extension_alembic_events()

Safe to call multiple times; subsequent calls are no-ops.

Return type:

None

Shared column-reference validation and marker helpers.

All codegen_database constraint types (CodegenDatabaseCheck, CodegenDatabaseIndex) use {column_name} markers. This module provides the regex, extraction, resolution, and validation helpers they share.

COLUMN_MARKER_RE = re.compile('\\{(\\w+)\\}')

Regex matching {column_name} markers in expressions.

extract_column_names(expression)

Extract {name} markers from an expression.

Parameters:

expression (str) – String containing {column_name} markers.

Return type:

list[str]

Returns:

Column names in order of first appearance, deduplicated.

resolve_markers(expression, mapping)

Replace each {col} marker with mapping(col).

Parameters:

• expression (str) – String containing {column_name} markers.

• mapping (Callable[[str], str]) – Callable that maps column names to their resolved form (e.g. identity for table-level, or lambda c: f"NEW.{c}" for triggers).

Return type:

str

Returns:

The expression with all markers replaced.

validate_column_references(label, columns, known_columns)

Raise if any column is not in known_columns.

Shared by check, index, and foreign-key plugins to ensure that user-provided column names actually exist on the target table or view.

Parameters:

• label (str) – Human-readable name for error messages (e.g. "CodegenDatabaseCheck 'pos_price'").

• columns (list[str]) – Column names to validate.

• known_columns (set[str]) – Set of known column names.

Raises:

CodegenDatabaseValidationError – If a column is not in known_columns.

Return type:

None

Cave exception types.

exception CodegenDatabaseValidationError

Raised when a codegen_database validation check fails.

Schema item validators for dimension factories.

frozen_function_default(server_default)

Return the SQL of a server_default that PostgreSQL will freeze.

A server_default given as a plain str is emitted as a quoted SQL literal (DEFAULT 'now()'), which PostgreSQL evaluates once at DDL time – every row then shares that single frozen value. When the string is shaped like a function call (now(), gen_random_uuid(), nextval(...) …) that is almost never what was meant. Wrap it in func.<fn>() or text(...) so it renders as a live call re-evaluated per row.

Returns the offending string, or None when the default is safe (a SQL element / text() clause, or a plain literal constant such as "0" / "true").

Return type:

str | None

is_schema_item_not_primary_key(item)

Return True if the item is not a primary key column.

Parameters:

item (SchemaItem | CodegenDatabaseCheck | CodegenDatabaseIndex) – The schema item to inspect.

Return type:

bool

Returns:

True if item is not a primary key column.

reject_frozen_function_defaults(tables)

Raise if any column defaults a function call passed as a plain string.

Guards against the silent DEFAULT 'now()' bug (frozen at migration time) across every column in tables – both user-supplied columns and the timestamp columns the built-in plugins inject.

Return type:

None

validate_schema_items(items, *, validators=None)

Validate a list of SchemaItems against the given validators.

Parameters:

• items (list) – Schema items to validate.

• validators (list[_SchemaItemValidator] | None) – Validators to run; defaults to [is_schema_item_not_primary_key].

Raises:

CodegenDatabaseValidationError – If any item fails a validator.

Return type:

None

Factory internals

The resource factories that plugins compose into. Most users interact with these through the built-in plugins rather than directly.

Core ResourceFactory: plugin runner.

class ResourceFactory(tablename, schemaname, metadata, schema_items, *, config=None, plugins=None, extra_plugins=None)

Core factory: resolves plugins and runs them in dependency order.

Subclasses declare DEFAULT_PLUGINS for user-facing defaults and _INTERNAL_PLUGINS for always-present built-in logic. Callers can override or extend the plugin list via plugins / extra_plugins, and inject global plugins via config.

Plugin execution order is determined by each plugin’s produces and requires declarations. Plugins with no declared dependencies run in the order they appear in the list.

Resolution order: global_plugins + user_plugins + internal_plugins, then topological sort.

If no user or global plugin produces pk_columns and internal plugins are present, a SerialPKPlugin is auto-prepended.

Parameters:

• tablename (str) – Name of the dimension table.

• schemaname (str) – PostgreSQL schema for all generated objects.

• metadata (MetaData) – SQLAlchemy MetaData the objects are bound to.

• schema_items (list[SchemaItem | CodegenDatabaseCheck | CodegenDatabaseIndex]) – Column and constraint definitions. Must not include a primary key column.

• config (object | None) – Optional global config supplying prepended plugins.

• plugins (list[Plugin | PluginCollection] | None) – If given, replaces DEFAULT_PLUGINS entirely.

• extra_plugins (list[Plugin | PluginCollection] | None) – Appended to the resolved plugin list.

Raises:

CodegenDatabaseValidationError – If any schema item fails validation, two plugins share a singleton group, two plugins produce the same ctx key, or a plugin dependency cycle is detected.

ctx: FactoryContext

The factory context after plugin execution.

Downstream view factories and query builders read this to access tables, columns, and other plugin outputs.

table: FromClause

The root selectable created by the factory.

This is the __root__ context value set by the table plugin, exposing the column metadata for use in queries, foreign key references, and ledger event lambdas.

Factory context dataclass.

class ContextSource(*args, **kwargs)

Structural type for objects that expose a factory context.

Both ResourceFactory instances and CodegenDatabaseBase / CodegenDatabaseView subclasses satisfy this protocol — the latter as class objects (ctx is a class attribute set by the declarative base’s class-construction hook).

Query builders (construct_ledger_balance_query(), ledger_event_function()) accept any ContextSource, making imperative and declarative styles interchangeable.

class FactoryContext(tablename, schemaname, metadata, schema_items, plugins)

Carries inputs and accumulates plugin outputs.

Typed input fields (set by the factory, read-only for plugins):

• tablename, schemaname, metadata, schema_items, plugins – as passed to the factory constructor.

Plugin store (read/write via item syntax):

Plugins communicate by storing and retrieving arbitrary values using string keys. The key names a plugin reads and writes are explicit constructor arguments on that plugin (with sensible defaults), so multiple independent pipelines can coexist by using distinct keys.

ctx["key"] = value

Store a value. Raises KeyError if key is already set – two plugins writing the same key is almost certainly a mistake. Use ctx.set("key", value, force=True) to override intentionally.

ctx["key"]

Retrieve a value. Raises KeyError with a plugin-ordering hint when the key is absent.

"key" in ctx

Test whether a key has been set without raising.

Injected columns (append-only list):

Plugins that provide columns for table construction (e.g. CreatedAtPlugin, UUIDEntryIDPlugin, DoubleEntryPlugin) append Column objects to ctx.injected_columns. Table plugins spread this list into the table definition alongside PK and dimension columns.

property columns: list[Column]

Return only Column instances from schema_items.

Useful when a plugin needs to iterate over column definitions (e.g. to extract column names or types).

property dim_column_names: list[str]

Return writable (non-PK, non-computed) column names.

Filters out primary-key and computed columns from schema_items, leaving only the user-defined dimension columns that a plugin should read or write. Equivalent to the _dim_column_names helper that was previously duplicated across multiple plugin modules.

get(key, default=None)

Same as underlying dict.get()

Return type:

Any

property pk_column_name: str

Return the primary key column name.

Shorthand for ctx["pk_columns"].first_key. Requires a PK plugin (e.g. SerialPKPlugin) to have run first.

Raises:

KeyError – If pk_columns has not been set yet.

set(key, value, *, force=False)

Store value under key, with optional override.

Parameters:

• key (str) – The store key to write.

• value (Any) – The value to store.

• force (bool) – If True, overwrite an existing value without raising. Use this when a plugin intentionally replaces a previous plugin’s output.

Raises:

KeyError – If key is already set and force is False.

Return type:

None

setdefault(key, value)

Same as underlying setdefault

Return type:

Any

property table_items: list[SchemaItem]

Return schema items suitable for table creation.

Filters out CodegenDatabaseCheck (which are handled by dedicated check plugins) but keeps all real SQLAlchemy SchemaItem objects: columns, constraints, indexes, computed columns, etc.

Simple dimension resource factory.

class CodegenDatabaseSimple(tablename, schemaname, metadata, schema_items, *, config=None, plugins=None, extra_plugins=None)

Create a simple dimension: one table with optional checks.

Internal plugins (always present):

1. SimpleTablePlugin – raw backing table ({tablename}_raw).

2. SimpleViewPlugin – writable view ({tablename}).

3. TableCheckPlugin – check constraints on the raw table.

4. TableIndexPlugin – indexes on the raw table.

5. TableFKPlugin – foreign keys on the raw table.

6. RawTableProtectionPlugin – blocks direct DML on the raw table.

7. InsteadOfTriggerPlugin – INSTEAD OF triggers on the dimension view.

A SerialPKPlugin is auto-added when no user plugin produces pk_columns.

Parameters:

• tablename (str) – Name of the dimension table.

• schemaname (str) – PostgreSQL schema for generated objects.

• metadata (MetaData) – SQLAlchemy MetaData to register on.

• schema_items (list[SchemaItem | CodegenDatabaseCheck | CodegenDatabaseIndex]) – Column and constraint definitions.

• plugins (list[Plugin | PluginCollection] | None) – Behaviour-modifying plugins (e.g. UUIDV4PKPlugin).

• extra_plugins (list[Plugin | PluginCollection] | None) – Appended to the resolved plugin list.

class SimpleTablePlugin

Create a single backing table for a simple dimension.

Creates {tablename}_raw and stores it in ctx["raw_table"]. The writable view {tablename} is created by SimpleViewPlugin().

run(ctx)

Create the raw table and store it in ctx.

Return type:

None

SimpleViewPlugin()

Create a configured ViewPlugin for simple dimensions.

Registers {tablename} as a view over {tablename}_raw and stores the proxy in ctx["primary"].

Return type:

ViewPlugin

Returns:

A ViewPlugin configured for simple passthrough views.

Append-only dimension resource factory.

class AppendOnlyTablePlugin(root_key='root_table', attributes_key='attributes')

Create the root and attributes tables for an append-only dim.

Parameters:

• root_key (str) – Key in ctx for the entity root table (default "root_table").

• attributes_key (str) – Key in ctx for the append-only attributes log (default "attributes").

run(ctx)

Create root and attributes tables.

Return type:

None

AppendOnlyViewPlugin(root_key='root_table', attributes_key='attributes', primary_key='primary')

Create a configured ViewPlugin for append-only dimensions.

Parameters:

• root_key (str) – Key in ctx for the entity root table (default "root_table").

• attributes_key (str) – Key in ctx for the attributes log (default "attributes").

• primary_key (str) – Key in ctx to store the view proxy under (default "primary").

Return type:

ViewPlugin

Returns:

A ViewPlugin configured for append-only join views.

class CodegenDatabaseAppendOnly(tablename, schemaname, metadata, schema_items, *, config=None, plugins=None, extra_plugins=None)

Create an append-only (SCD Type 2) dimension.

Internal plugins (always present), in order:

The column-name plugin construct_column_name_plugin() sets the created_at column name. Then:

1. UniqueColumnCheckPlugin – lifts column-level unique=True into INSTEAD OF trigger checks (must run before the table is built so the unique flag can be stripped in time).

2. AppendOnlyTablePlugin – root + attributes tables.

3. AppendOnlyViewPlugin – join view proxy.

4. TableIndexPlugin – indices on the attributes table.

5. TableFKPlugin – foreign keys on the attributes table.

6. InsteadOfTriggerPlugin – INSTEAD OF triggers (activates when a view plugin produces "primary").

TableCheckPlugin is auto-added by the base factory when not already present.

A SerialPKPlugin is auto-added when no user plugin produces pk_columns.

class UniqueColumnCheckPlugin

Lift column-level unique=True into INSTEAD OF trigger checks.

SQLAlchemy column-level uniqueness propagates onto the underlying attributes log. An append-only UPDATE rewrites into an INSERT that carries every column value forward, so the new revision collides with its own predecessor on any unique column even when the value is unchanged.

For each column declared with unique=True this plugin:

1. Strips the unique flag before AppendOnlyTablePlugin attaches the column to the attributes table – once attached, SQLAlchemy materialises a UniqueConstraint eagerly and clearing the flag no longer suppresses it.

2. Registers an INSTEAD OF trigger on the public view that rejects duplicates among the current revisions only.

The trigger is best-effort: it does not take a row lock, so two concurrent inserts can both pass the check before either commits. Use SERIALIZABLE isolation or an explicit advisory lock if airtight uniqueness is required.

run(ctx)

Strip unique=True and register trigger-based checks.

Return type:

None

EAV dimension resource factory.

class CodegenDatabaseEAV(tablename, schemaname, metadata, schema_items, *, config=None, plugins=None, extra_plugins=None)

Create an EAV (Entity-Attribute-Value) dimension.

Internal plugins (always present), in order:

The column-name plugin construct_column_name_plugin() sets the created_at column name. Then:

1. EAVTablePlugin – entity + attribute tables.

2. EAVViewPlugin – pivot view proxy.

3. TriggerCheckPlugin – trigger-based checks on the pivot view.

4. InsteadOfTriggerPlugin – INSTEAD OF triggers (activates when a view plugin produces "primary").

A SerialPKPlugin is auto-added when no user plugin produces pk_columns.

class EAVTablePlugin(entity_key='entity', attribute_key='attribute', mappings_key='eav_mappings')

Create entity and attribute tables for an EAV dimension.

Parameters:

• entity_key (str) – Key in ctx for the entity root table (default "entity").

• attribute_key (str) – Key in ctx for the attribute log (default "attribute").

• mappings_key (str) – Key in ctx for the EAV mappings list, shared with the view and trigger plugins (default "eav_mappings").

run(ctx)

Create entity and attribute tables.

Return type:

None

EAVViewPlugin(entity_key='entity', attribute_key='attribute', mappings_key='eav_mappings', primary_key='primary')

Create a configured ViewPlugin for EAV dimensions.

Parameters:

• entity_key (str) – Key in ctx for the entity root table (default "entity").

• attribute_key (str) – Key in ctx for the attribute log (default "attribute").

• mappings_key (str) – Key in ctx for the EAV mappings list (default "eav_mappings").

• primary_key (str) – Key in ctx to store the view proxy under (default "primary").

Return type:

ViewPlugin

Returns:

A ViewPlugin configured for EAV pivot views.

class TriggerCheckPlugin(table_key='primary')

Enforce checks via INSTEAD OF triggers (EAV dimensions).

Generates a single trigger function per view per operation (INSERT/UPDATE) that validates all CodegenDatabaseCheck items. Triggers use a _check_ prefix to fire before the main dimension triggers (Postgres fires multiple INSTEAD OF triggers in alphabetical order by name).

Parameters:

table_key (str) – Key in ctx for the trigger target view (default "primary").

Ledger resource factory.

class CodegenDatabaseLedger(tablename, schemaname, metadata, schema_items, *, config=None, plugins=None, extra_plugins=None)

Create a ledger: append-only table with a value column.

Internal plugins (always present), in order:

1. UUIDEntryIDPlugin – UUID entry ID for correlating related entries.

Next, the column-name plugin construct_column_name_plugin() sets the created_at column name. Then:

2. LedgerTablePlugin – raw backing table ({tablename}_raw).

3. LedgerViewPlugin – writable view ({tablename}).

4. RawTableProtectionPlugin – blocks direct DML on the raw table.

5. InsteadOfTriggerPlugin – INSTEAD OF triggers on the dimension view.

A SerialPKPlugin is auto-added when no user plugin produces pk_columns.

Use construct_ledger_balance_query(), construct_ledger_latest_query(), and ledger_event_function() with CodegenDatabaseFunction for derived views and event functions.

class LedgerTablePlugin(value_type='integer', *, precision=None, scale=None)

Create a ledger raw table with a value column.

Combines ctx["pk_columns"], ctx.injected_columns (provided by upstream plugins like UUIDEntryIDPlugin, CreatedAtPlugin, and DoubleEntryPlugin), a value column, and ctx.table_items (dimension columns) into a single append-only table named {tablename}_raw.

The writable view {tablename} is created by LedgerViewPlugin().

Parameters:

• value_type (Literal['integer', 'numeric', 'decimal']) – Type for the value column. One of "integer", "numeric", or "decimal" (default "integer").

• precision (int | None) – Total number of digits when value_type is "numeric" / "decimal" – the NUMERIC(precision, scale) first argument. None leaves the column unconstrained.

• scale (int | None) – Digits after the decimal point. Requires precision.

Raises:

CodegenDatabaseValidationError – If value_type is not a recognised type, or precision/scale are misconfigured.

run(ctx)

Create the ledger raw table and store it in ctx.

Return type:

None

LedgerViewPlugin()

Create a configured ViewPlugin for ledger dimensions.

Registers {tablename} as a view over {tablename}_raw and stores the proxy in ctx["primary"].

Return type:

ViewPlugin

Returns:

A ViewPlugin configured for ledger passthrough views.

class UUIDEntryIDPlugin(column_name='entry_id')

Provide a UUIDv4 entry ID column for ledger tables.

Stores a Column in ctx["entry_id_column"] that downstream table plugins splice into the table definition. The column uses PostgreSQL’s gen_random_uuid() as a server default so callers can omit it for single-entry inserts while still providing an explicit value to correlate multi-row entries.

Parameters:

column_name (str) – Name of the entry ID column (default "entry_id").

run(ctx)

Store the entry ID column and inject it.

Return type:

None

Built-in plugins

Plugins that ship with codegen_database and power the default factories.

Primary key plugins.

class SerialPKPlugin(column_name='id')

Provide an auto-increment integer primary key column.

Parameters:

column_name (str) – Name of the PK column (default "id").

run(ctx)

Store a PrimaryKeyColumns in the ctx store.

Return type:

None

class UUIDV4PKPlugin(column_name='id')

Provide a UUIDv4 primary key column.

Uses PostgreSQL’s gen_random_uuid() as the server default so rows get a unique identifier without client-side generation.

Parameters:

column_name (str) – Name of the PK column (default "id").

run(ctx)

Store a PrimaryKeyColumns in the ctx store.

Return type:

None

class UUIDV7PKPlugin(column_name='id')

Provide a UUIDv7 primary key column.

Uses PostgreSQL 18’s uuidv7() as the server default to generate time-ordered UUIDs. These sort chronologically, making them friendlier to B-tree indexes than random UUIDv4 values.

Requires PostgreSQL 18 or later (declared via @requires(MinPGVersion(18))). Use check_pg_version() to validate the server version before applying DDL.

Parameters:

column_name (str) – Name of the PK column (default "id").

run(ctx)

Store a PrimaryKeyColumns in the ctx store.

Return type:

None

Column name registry plugin.

construct_column_name_plugin(ctx_key, column_name)

Create a column-name-registry plugin for ctx_key.

Parameters:

• ctx_key (str) – The ctx key under which column_name is stored.

• column_name (str) – Name of the column.

Return type:

Plugin

Returns:

A plugin instance that publishes column_name under ctx_key at factory run time.

Check constraint plugins for codegen_database dimensions.

TableCheckPlugin converts CodegenDatabaseCheck items into real SQLAlchemy CheckConstraint objects on a table (for simple and append-only dimensions).

class TableCheckPlugin(table_key='primary')

Materialize check items as table constraints.

Turns CodegenDatabaseCheck items into SQLAlchemy CheckConstraint objects.

Reads CodegenDatabaseCheck items from ctx.schema_items, resolves {col} markers to plain column names (identity), and appends real CheckConstraint objects to the target table.

Parameters:

table_key (str) – Key in ctx for the target table (default "primary").

Index plugin for codegen_database dimensions.

TableIndexPlugin converts CodegenDatabaseIndex items into real SQLAlchemy Index objects on a table.

class TableIndexPlugin(table_key='primary')

Materialize index items as table indexes.

Turns CodegenDatabaseIndex items into SQLAlchemy Index objects.

Reads CodegenDatabaseIndex items from ctx.schema_items, validates column names, and creates Index objects on the target table. Extra keyword arguments on each CodegenDatabaseIndex are passed through to the underlying sqlalchemy.Index.

Parameters:

table_key (str) – Key in ctx for the target table (default "primary").

run(ctx)

Collect, validate, and create indexes.

Return type:

None

Foreign key plugin for codegen_database dimensions.

TableFKPlugin converts inline CodegenDatabaseForeignKey column markers into real SQLAlchemy ForeignKeyConstraint objects on a table. Two-part references ("dimension.column") are resolved via the dimension registry in metadata.info.

class TableFKPlugin(table_key='primary')

Materialize FK declarations as ForeignKeyConstraint objects.

Handles inline CodegenDatabaseForeignKey markers attached to Column constructors (single-column).

Two-part "dimension.column" references are resolved via the dimension registry. Factories run at class-creation time, so a reference may name a dimension whose model hasn’t been imported yet – those are deferred and materialize when the dimension registers (see register_dimension()); import order never constrains who may reference whom. Three-part "schema.table.column" references are passed through directly.

Parameters:

table_key (str) – Key in ctx for the target table (default "primary").

run(ctx)

Collect and create foreign key constraints from inline markers.

Return type:

None

User-facing plugins for ledger (append-only value) tables.

class DoubleEntryPlugin(column_name='direction')

Add debit/credit semantics to a ledger table.

Adds a direction column ('debit' or 'credit') to the schema items so that LedgerTablePlugin includes it in the table. Also registers an AFTER INSERT constraint trigger that validates all rows sharing an entry_id have equal total debits and credits.

This plugin must appear before LedgerTablePlugin in the plugin list so its column is included in the table definition.

Parameters:

column_name (str) – Name of the direction column (default "direction").

run(ctx)

Inject the direction column and store its name.

Return type:

None

class DoubleEntryTriggerPlugin(table_key='raw_table')

Register an AFTER INSERT trigger enforcing balanced entries.

Validates that for every entry_id in the inserted batch, the sum of debit values equals the sum of credit values. Raises a PostgreSQL exception if any entry is unbalanced.

Uses a statement-level trigger with a REFERENCING NEW TABLE transition table so that multi-row inserts are checked as a whole, not row-by-row.

Must run after LedgerTablePlugin (needs the table) and after DoubleEntryPlugin (needs column name).

Parameters:

table_key (str) – Key in ctx for the backing table (default "primary").

run(ctx)

Register the constraint trigger on the ledger table.

Return type:

None

class LedgerBalanceCheckPlugin(dimensions, min_balance=0, table_key='raw_table')

Enforce a minimum balance per dimension group.

Registers an AFTER INSERT FOR EACH STATEMENT trigger that checks SUM(value) >= min_balance for every dimension group affected by the inserted rows. If any group violates the constraint the entire statement is rejected.

Uses the same REFERENCING NEW TABLE transition-table pattern as DoubleEntryTriggerPlugin.

Parameters:

• dimensions (list[str]) – Column names that define a balance group. Must be a non-empty list.

• min_balance (int) – The minimum allowed SUM(value) per group (default 0).

• table_key (str) – Key in ctx for the backing table (default "primary").

Raises:

CodegenDatabaseValidationError – If dimensions is empty.

run(ctx)

Register the balance-check trigger.

Return type:

None

Plugin that prevents direct DML on raw backing tables.

Direct INSERT/UPDATE/DELETE on raw backing tables bypasses the INSTEAD OF triggers on the dimension views, which can corrupt dimension state (e.g. breaking SCD Type 2 history in append-only dimensions, leaving orphaned EAV rows).

RawTableProtectionPlugin installs BEFORE triggers on every raw table it is given. The triggers raise an exception unless one of two conditions holds: the insert arrives via the INSTEAD OF trigger on the dimension view (pg_trigger_depth() >= 2), or the caller has set the transaction-local config parameter codegen_database.event_fn_active = 'true' via set_config('codegen_database.event_fn_active', 'true', true). Trusted event functions such as those built by ledger_event_function() set this flag so they can insert directly into the raw table in a single batched statement, which is required for correct double-entry validation.

class RawTableProtectionPlugin(*table_keys)

Prevent direct DML on raw backing tables.

Installs BEFORE INSERT/UPDATE/DELETE triggers on every raw table specified by table_keys. The triggers raise an exception when called at trigger depth 0 (i.e. directly, not from within another trigger), so mutations must go through the dimension view.

All mutations through the dimension view arrive via an INSTEAD OF trigger at depth >= 1, which the protection triggers allow through.

Parameters:

*table_keys (str) – One or more ctx keys whose values are the raw Table objects to protect.

Example:

RawTableProtectionPlugin("root_table", "attributes")

resolved_requires()

Return the ctx keys this plugin reads.

Overrides the base implementation so that the topological sort correctly places this plugin after all table-creating plugins.

Return type:

list[str]

run(ctx)

Register protection triggers on each raw backing table.

Return type:

None

Soft-delete plugin for codegen_database dimensions.

Intercepts the INSTEAD OF DELETE trigger on the dimension view and converts it into a timestamp-based soft-delete. The dimension view is filtered to hide soft-deleted rows, keeping the interface clean.

The plugin itself is factory-agnostic: it only records intent in the context (the soft-delete column name and the delete-trigger template to use). Each factory’s table plugin injects the deleted_at column on the correct backing table, and each factory’s delete template performs the timestamp update. The exact behaviour per factory type:

• Simple (table_key="raw_table"): UPDATE raw_table SET deleted_at = now() WHERE id = OLD.id. The view appends WHERE deleted_at IS NULL.

• AppendOnly (table_key="attributes"): inserts a new attributes version with deleted_at = now() and updates the root FK pointer. The view appends WHERE deleted_at IS NULL.

• EAV (table_key="entity"): UPDATE entity SET deleted_at = now() WHERE id = OLD.id. The deleted_at column lives on the auto-generated entity table and the pivot view filters it in-query (a plain WHERE append would land after the pivot’s GROUP BY), so the view-patching step is skipped.

• Ledger: not applicable – ledger rows are immutable by design.

Usage:

# Simple dimension
class Users(Base):
    __tablename__ = "users"
    __table_args__ = {"schema": "public"}
    __plugins__ = [SoftDeletePlugin()]

    name = Column(String, nullable=False)

# AppendOnly dimension
class Students(Base):
    __tablename__ = "students"
    __table_args__ = {"schema": "public"}
    __factory__ = CodegenDatabaseAppendOnly
    __plugins__ = [SoftDeletePlugin(table_key="attributes")]

    name = Column(String, nullable=False)

# EAV dimension
class Products(Base):
    __tablename__ = "products"
    __table_args__ = {"schema": "public"}
    __factory__ = CodegenDatabaseEAV
    __plugins__ = [SoftDeletePlugin(table_key="entity")]

    sku = Column(String, nullable=False)

The deleted_at column is injected automatically onto the backing table – you do not need to declare it on the model. For Simple and AppendOnly dimensions you may declare it explicitly (as a nullable=True DateTime column) if you want it visible on the model; the injection is skipped when the column already exists. For EAV you must not declare it: user columns become EAV attributes, and the soft-delete flag must live on the entity table instead.

After soft-deletion the row remains in the backing table with deleted_at set. The view {tablename} only shows rows where deleted_at IS NULL.

class SoftDeletePlugin(column_name='deleted_at', table_key='raw_table')

Soft delete plugin.

Parameters:

• column_name (str) – Name of the soft-delete timestamp column (default "deleted_at").

• table_key (str) – Backing table the soft-delete column lives on: "raw_table" for Simple (default), "attributes" for AppendOnly, "entity" for EAV.

class SoftDeletePluginAfterRoot(column_name='deleted_at', table_key='raw_table')

Patch the view to filter out soft-deleted rows.

run(ctx)

Append WHERE deleted_at IS NULL to the registered view.

Skipped when the factory already filters soft-deleted rows inside its view query (it sets soft_delete_view_filtered). A blind WHERE append cannot be used there – e.g. EAV’s pivot view ends in GROUP BY and the clause would be misplaced.

Return type:

None

class SoftDeletePluginBeforeRoot(column_name='deleted_at')

Record soft-delete intent for the factory’s table/trigger plugins.

Stores the soft-delete column name under deleted_at_column (read by the factory’s table plugin to create the column and by EAV’s view builder to filter the pivot) and injects a DeleteTriggerOverride so the factory’s InsteadOfTriggerPlugin renders a soft-delete body instead of a physical DELETE – without the factory’s ops builder needing any soft-delete-specific code. This plugin runs before the table plugin (which declares deleted_at_column in its requires).

Parameters:

column_name (str) – Name of the soft-delete timestamp column (default "deleted_at").

run(ctx)

Record the soft-delete column and delete override in ctx.

Return type:

None

soft_delete_columns(ctx)

Return the soft-delete column(s) to inject onto a backing table.

Factory table plugins splat this into their Table(...) call so the deleted_at column is created whenever a SoftDeletePlugin recorded its intent in ctx.

Returns an empty list when no soft-delete is configured, or when the user already declared the column among schema_items (Simple and AppendOnly allow that), so it is safe to call unconditionally.

Parameters:

ctx (FactoryContext) – The factory context.

Return type:

list[Column]

Returns:

A one-element list with the nullable DateTime soft-delete column, or an empty list.

Row-level security plugins for codegen_database dimensions.

Each plugin registers one or more PostgreSQL RLS policies on the raw backing table so that queries are automatically filtered by the current session’s app.* settings.

Four pre-built policies cover the most common patterns:

• OwnerRLSPlugin – generic: any column matched against any app.* setting.

• UserRLSPlugin – per-user isolation (app.user_id).

• TenantRLSPlugin – per-tenant isolation (app.tenant_id).

• TenantUserRLSPlugin – tenant isolation AND per-user visibility within the tenant.

All plugins accept an optional bypass_roles list. Roles in that list receive an unconditional USING (true) policy so they can read and write every row – useful for service accounts and trusted roles.

Prerequisites

1. RLS policy diffing in Alembic requires calling register_rls_alembic_events() once in env.py.

2. The application must call SET LOCAL app.user_id = '...' (or equivalent) for each request so the session setting is available.

Usage:

class Users(Base):
    __tablename__ = "users"
    __table_args__ = {"schema": "public"}
    __plugins__ = [UserRLSPlugin(bypass_roles=["authenticator"])]

    user_id = Column(String, nullable=False)
    name = Column(String, nullable=False)

class OwnerRLSPlugin(column, setting, *, policy_name='owner_policy', bypass_roles=None, table_key='raw_table')

Generic owner-column RLS: filter rows by a session setting.

Enables row-level security on the target table and creates a PERMISSIVE policy that restricts access to rows where current_setting(setting, true)::text = column::text.

Parameters:

• column (str) – Column that identifies the row’s owner.

• setting (str) – PostgreSQL session setting to compare against, e.g. "app.user_id".

• policy_name (str) – Name for the created policy (default "owner_policy").

• bypass_roles (list[str] | None) – Roles that receive an unconditional USING (true) policy, bypassing the owner filter.

• table_key (str) – Key in ctx for the table to protect (default "raw_table").

run(ctx)

Register the owner RLS policy on the backing table.

Return type:

None

class TenantRLSPlugin(tenant_column='tenant_id', *, bypass_roles=None, table_key='raw_table')

Per-tenant row isolation via app.tenant_id.

Filters rows to those where the nominated tenant_column matches current_setting('app.tenant_id', true).

Parameters:

• tenant_column (str) – Column holding the tenant ID (default "tenant_id").

• bypass_roles (list[str] | None) – Roles that bypass the filter.

• table_key (str) – Key in ctx for the backing table (default "raw_table").

run(ctx)

Register tenant-isolation RLS policy.

Return type:

None

class TenantUserRLSPlugin(tenant_column='tenant_id', user_column='user_id', *, bypass_roles=None, table_key='raw_table')

Tenant isolation combined with per-user visibility.

Rows are only visible when BOTH conditions hold:

• current_setting('app.tenant_id', true) matches tenant_column, AND

• current_setting('app.user_id', true) matches user_column.

Parameters:

• tenant_column (str) – Column holding the tenant ID (default "tenant_id").

• user_column (str) – Column holding the user ID (default "user_id").

• bypass_roles (list[str] | None) – Roles that bypass the filter.

• table_key (str) – Key in ctx for the backing table (default "raw_table").

run(ctx)

Register combined tenant+user RLS policy.

Return type:

None

class UserRLSPlugin(user_column='user_id', *, bypass_roles=None, table_key='raw_table')

Per-user row isolation via app.user_id.

Filters rows to those where the nominated user_column matches current_setting('app.user_id', true).

Parameters:

• user_column (str) – Column holding the owning user ID (default "user_id").

• bypass_roles (list[str] | None) – Roles that bypass the filter.

• table_key (str) – Key in ctx for the backing table (default "raw_table").

run(ctx)

Register user-isolation RLS policy.

Return type:

None

Temporal validity range plugin for codegen_database dimensions.

TemporalPlugin adds valid_from/valid_to columns to a dimension’s backing table and provides:

• A construct_temporal_as_of_query() helper that queries the table at a specific point in time.

• A GiST-based exclusion constraint (requiring the btree_gist PostgreSQL extension) to prevent overlapping validity ranges for the same entity.

The plugin works with any factory type that has a user-controlled backing table (CodegenDatabaseSimple, CodegenDatabaseAppendOnly, CodegenDatabaseLedger). The user supplies valid_from and valid_to columns in schema_items; the plugin validates them and adds the exclusion constraint.

Usage:

orders = CodegenDatabaseSimple(
    tablename="orders",
    schemaname="app",
    metadata=metadata,
    schema_items=[
        Column("order_id", Integer, nullable=False),
        Column("status", String, nullable=False),
        Column(
            "valid_from",
            DateTime(timezone=True),
            nullable=False,
            server_default=func.now(),
        ),
        Column("valid_to", DateTime(timezone=True), nullable=True),
    ],
    extra_plugins=[
        TemporalPlugin(
            subject_columns=["order_id"],
            valid_from_column="valid_from",
            valid_to_column="valid_to",
        ),
    ],
)

# Point-in-time query
query = construct_temporal_as_of_query(
    orders,
    as_of="2024-06-15T00:00:00Z",
    subject_columns=["order_id"],
)

The exclusion constraint prevents storing two overlapping validity windows for the same order_id. Requires btree_gist:

CREATE EXTENSION IF NOT EXISTS btree_gist;

class TemporalPlugin(subject_columns, valid_from_column='valid_from', valid_to_column='valid_to', table_key='raw_table')

Add temporal validity range support to a dimension.

Validates that valid_from_column and valid_to_column exist in the backing table, stores their names in ctx["temporal_columns"], and appends a GiST exclusion constraint to the table so that no two rows with the same subject_columns values have overlapping tstzrange\ s.

The exclusion constraint uses:

EXCLUDE USING GIST (
    col1 WITH =, ...,
    tstzrange(valid_from, COALESCE(valid_to, 'infinity'), '[)')
    WITH &&
)

which requires the btree_gist extension to be installed.

Parameters:

• subject_columns (list[str]) – Columns that identify a unique entity across multiple temporal versions (e.g. ["order_id"]).

• valid_from_column (str) – Name of the range-start column (default "valid_from").

• valid_to_column (str) – Name of the range-end column (default "valid_to").

• table_key (str) – Key in ctx for the backing table to add the constraint to (default "raw_table").

Raises:

CodegenDatabaseValidationError – If valid_from_column or valid_to_column is missing, non-nullable for valid_from (it must have a value), or not a DateTime column.

run(ctx)

Validate columns and add the exclusion constraint.

Return type:

None

construct_temporal_as_of_query(source, as_of, subject_columns=None, *, table_key='raw_table')

Build a point-in-time query for a temporal dimension.

Returns rows where:

valid_from <= :as_of AND (valid_to IS NULL OR valid_to > :as_of)

Parameters:

• source (ContextSource) – A factory instance or declarative class that was built with TemporalPlugin.

• as_of (str) – ISO 8601 timestamp string or PostgreSQL-parseable timestamp for the point in time to query.

• subject_columns (list[str] | None) – Columns identifying unique entities. When provided, applies DISTINCT ON to return only the most recent version per subject at as_of. When None, returns all matching rows.

• table_key (str) – Key in ctx for the backing table (default "raw_table").

Return type:

Select

Returns:

A SQLAlchemy Select.

Raises:

CodegenDatabaseValidationError – If TemporalPlugin has not been applied to source.

Ledger snapshot plugin: maintain a running-total table.

LedgerSnapshotPlugin creates a {tablename}_snapshot table whose rows always reflect SUM(value) per dimension group. An AFTER INSERT FOR EACH STATEMENT trigger on the ledger raw table applies each batch of inserts as incremental UPSERT operations, so the snapshot stays in sync without a full re-scan.

This is the performance-optimised alternative to construct_ledger_balance_query() when the ledger is large and balance reads are frequent.

Usage:

inventory = CodegenDatabaseLedger(
    tablename="inventory",
    schemaname="private",
    metadata=metadata,
    schema_items=[
        Column("warehouse", String, nullable=False),
        Column("sku", String, nullable=False),
    ],
    extra_plugins=[
        LedgerSnapshotPlugin(dimensions=["warehouse", "sku"]),
    ],
)

# inventory.snapshot_table is a joinable proxy.

After construction the snapshot_table attribute on the factory instance holds a joinable SQLAlchemy Table whose columns are the declared dimension columns plus balance (same type as value) and updated_at.

Note: the snapshot does not enforce a minimum balance. Use LedgerBalanceCheckPlugin on the ledger table for that.

class LedgerSnapshotPlugin(dimensions, value_type='integer', *, precision=None, scale=None, table_key='raw_table')

Maintain a running-total snapshot table for a ledger.

Creates {tablename}_snapshot with one row per unique combination of dimensions columns. An AFTER INSERT FOR EACH STATEMENT trigger incrementally applies each batch of ledger inserts via INSERT ... ON CONFLICT DO UPDATE, keeping the snapshot current without a full aggregation scan.

After the plugin runs, ctx["snapshot_table"] is a joinable SQLAlchemy Table whose columns are the declared dimensions plus balance and updated_at.

Parameters:

• dimensions (list[str]) – Column names to group the balance by. Must be a non-empty list matching columns on the ledger raw table.

• value_type (Literal['integer', 'numeric', 'decimal']) – Type for the balance column. One of "integer", "numeric", or "decimal". Should match the ledger’s value_type (default "integer").

• precision (int | None) – Total number of digits when value_type is "numeric" / "decimal". Should match the ledger’s precision. None leaves the column unconstrained.

• scale (int | None) – Digits after the decimal point. Requires precision.

• table_key (str) – Key in ctx for the ledger raw table (default "raw_table").

Raises:

CodegenDatabaseValidationError – If dimensions is empty, value_type is unrecognised, or precision/scale are misconfigured.

run(ctx)

Create the snapshot table and the upsert trigger.

Return type:

None

Search vector plugin for codegen_database dimensions.

SearchVectorPlugin maintains a tsvector column on the backing table, keeping it in sync with one or more source text columns via a BEFORE INSERT OR UPDATE trigger.

PostgreSQL’s full-text search infrastructure (tsvector, to_tsvector, @@, GIN indexes) is built into the database engine — no extension is required for basic usage.

Usage:

products = CodegenDatabaseSimple(
    "products",
    "app",
    metadata,
    schema_items=[
        Column("name", String, nullable=False),
        Column("description", String),
        Column("search_vector", TSVECTOR, nullable=True),
    ],
    extra_plugins=[
        SearchVectorPlugin(
            source_columns=["name", "description"],
            vector_column="search_vector",
        ),
    ],
)

The trigger function is registered under {schema}.{schema}_{tablename}_search_vector_update() and fires BEFORE INSERT OR UPDATE on the raw backing table.

You may optionally pass a GIN index via CodegenDatabaseIndex on the vector column (e.g. CodegenDatabaseIndex(columns=["search_vector"], using="gin")), which is the standard way to make full-text searches fast.

If the text search configuration (ts_config) references a dictionary installed via a PostgreSQL extension (e.g. unaccent, pg_trgm), register that extension in metadata via register_pg_extension().

class SearchVectorPlugin(source_columns, vector_column='search_vector', ts_config='english', table_key='raw_table')

Maintain a tsvector column via a BEFORE trigger.

Generates and registers a BEFORE INSERT OR UPDATE trigger function that recomputes:

NEW.vector_column := to_tsvector(
    ts_config,
    COALESCE(col1, '') || ' ' || COALESCE(col2, '') || ...
)

Source columns are concatenated with a space separator; NULL values are coalesced to an empty string so they do not suppress the entire vector.

Parameters:

• source_columns (list[str]) – One or more column names whose text content feeds the search vector. At least one required.

• vector_column (str) – Name of the tsvector column to maintain (default "search_vector"). Must exist in schema_items with a compatible type.

• ts_config (str) – PostgreSQL text-search configuration name (default "english"). Use any configuration installed in the database. If the configuration depends on a PostgreSQL extension (e.g. "unaccent"), register that extension with register_pg_extension().

• table_key (str) – Key in ctx for the backing table to attach the trigger to (default "raw_table").

Raises:

CodegenDatabaseValidationError – If source_columns is empty, any source column is absent from the table, or vector_column is absent from the table.

run(ctx)

Validate columns, then register the trigger function.

Return type:

None

Generic view plugin for dimension factories.

class ViewPlugin(query_builder, proxy_builder, primary_key='primary', extra_requires=None)

Register a view and store a proxy table in ctx.

A generic, composable view plugin. Each factory type provides a query_builder that returns the view SQL and a proxy_builder that returns proxy columns for downstream plugins.

Parameters:

• query_builder (Callable[[FactoryContext], str]) – Callable (ctx) -> str returning the compiled SQL for the view definition.

• proxy_builder (Callable[[FactoryContext], list[Column]]) – Callable (ctx) -> list[Column] returning proxy columns for the view’s selectable proxy.

• primary_key (str) – Key in ctx to store the view proxy under (default "primary").

• extra_requires (list[str] | None) – Additional ctx keys this plugin depends on. Declared to the topological sorter so that upstream plugins run first.

resolved_requires()

Return base requires plus extra runtime keys.

Return type:

list[str]

run(ctx)

Register the view and store the proxy in ctx.

Return type:

None

Generic INSTEAD OF trigger plugin.

class DeleteTriggerOverride(template_name, template_vars=<factory>)

Replacement template for a dimension’s INSTEAD OF DELETE body.

A plugin (e.g. SoftDeletePlugin) injects this into the factory context under DELETE_TRIGGER_OVERRIDE_KEY to swap the default physical-delete template for an alternative (e.g. a soft-delete UPDATE). Each factory renders its delete op through render_delete_op(), so no factory needs to special-case the override itself.

Parameters:

• template_name (str) – Filename of the delete template to render instead of delete.plpgsql.mako, resolved within the calling factory’s own template directory.

• template_vars (dict[str, Any]) – Extra template variables merged on top of the factory’s base delete vars (e.g. deleted_at_column).

class InsteadOfTriggerPlugin(ops_builder, naming_defaults, function_key, trigger_key, view_key='primary', permitted_operations=None, *, extra_requires=None)

Register INSTEAD OF triggers from pre-rendered PL/pgSQL bodies.

A generic, composable trigger plugin. Each factory type provides an ops_builder callable that reads from the factory context and returns a list of TriggerOp with fully rendered PL/pgSQL.

Parameters:

• ops_builder (Callable[[FactoryContext], list[TriggerOp]]) – Callable that takes a FactoryContext and returns a list of TriggerOp.

• naming_defaults (dict[str, str]) – Default naming templates for function and trigger names.

• function_key (str) – Key for function name resolution.

• trigger_key (str) – Key for trigger name resolution.

• view_key (str) – Key in ctx for the trigger target view. If absent from ctx, trigger registration is skipped.

• permitted_operations (list[str] | None) – When set, only operations whose names appear in this list are registered.

• extra_requires (list[str] | None) – Additional ctx keys to declare as dependencies for topological ordering.

resolved_requires()

Return base requires plus extra runtime keys.

Return type:

list[str]

run(ctx)

Register INSTEAD OF triggers on the target view.

Skips entirely when view_key is absent from ctx. When permitted_operations was not set at construction time, falls back to ctx["permitted_operations"] if present.

Return type:

None

class TriggerOp(name, body)

A single INSTEAD OF trigger operation.

Parameters:

• name (str) – DML operation name ("insert", "update", "delete").

• body (str) – Pre-rendered PL/pgSQL function body.

render_delete_op(ctx, templates_dir, base_vars)

Render the DELETE TriggerOp, honoring a context override.

Renders delete.plpgsql.mako from templates_dir with base_vars unless a plugin injected a DeleteTriggerOverride in ctx, in which case the override’s template and extra vars take precedence. This keeps soft-delete (and any future delete variant) out of each factory’s ops builder.

Parameters:

• ctx (FactoryContext) – The factory context.

• templates_dir (Path) – The calling factory’s template directory.

• base_vars (dict[str, Any]) – The factory’s default delete template variables.

Return type:

TriggerOp

Returns:

The rendered "delete" TriggerOp.

Utilities

compile_query(query)

Compile a SQLAlchemy query to a PostgreSQL SQL string.

Return type:

str

Migration glue (codegen_database.alembic.*)

Hooks your project’s env.py wires into Alembic. The comparator/renderer/rewriter modules are internal but documented for reference.

alembic_hook(config=None)

Register codegen_database’s alembic extensions.

Call before importing models.

Usage in env.py:

from codegen_database.alembic.register import (
    alembic_hook,
    configure_metadata,
    process_revision_directives,
)

alembic_hook()

# ... import models / build metadata ...

configure_metadata(target_metadata)

Then pass process_revision_directives to context.configure(process_revision_directives=...).

Parameters:

config (CodegenDatabaseConfig | None) – Optional config providing extensions whose configure_alembic() hooks will be called.

Return type:

None

configure_metadata(metadata, config=None)

Register schemas and extension hooks on metadata.

Parameters:

• metadata (MetaData) – The SQLAlchemy MetaData to configure.

• config (CodegenDatabaseConfig | None) – Optional config providing extensions. If None, falls back to metadata.info["codegen_database_config"].

Raises:

CodegenDatabaseValidationError – A two-part FK reference names a dimension no imported model registered.

Return type:

None

render_item(type_, obj, autogen_context)

Render custom column types for migrations.

The in-scope types (codegen_database.types plus sqlalchemy_utils encrypted types, e.g. behind fsh_lib.oauth’s token columns) are TypeDecorator wrappers around plain SQL types, carrying Python-side state a migration can neither render nor use – an enum class, an encryption-key callable. Alembic’s default rendering emits their dotted class name with no import, producing a migration that doesn’t run. Migrations only need the database-level type, so this renders each type’s own impl (TextEnum -> sa.Text(), EncryptedText -> sa.Text(), …) instead of maintaining a parallel name -> DDL table.

Pass this as render_item to context.configure(...) in your Alembic env.py.

Returns False for unrecognized objects so Alembic falls through to its default rendering.

Return type:

str | Literal[False]

class EntityIdentifier(schema='public', name=None, phase=None)

Identifies a database entity within a migration’s dependency graph.

name is None for schema-level entities (i.e. the entity is the schema). For tables, views, and functions, name holds the unqualified object name and schema holds its containing schema.

phase distinguishes drop and create ops for the same entity when an Update*Op has been expanded. "drop" ops are ordered before "create" ops for the same entity.

build_fk_graph_from_metadata(metadata)

Build a table FK dependency map from SQLAlchemy metadata.

Alembic’s DropTableOp carries only the table name and schema — no column or foreign key information. Without an external FK graph the topological sort cannot order drop operations correctly, leading to constraint violations when a referenced table is dropped before its dependents.

This function reads FK relationships from SQLAlchemy’s MetaData (which has the full schema) so that sort_migration_ops can order both create and drop operations safely.

Return type:

dict[tuple[str, str], set[tuple[str, str]]]

drop_spurious_declarative_updates(ops)

Drop no-op function/view/trigger Update ops (default-schema and renderer-whitespace churn). Real changes survive; dependent recreation a real change or column alter needs is re-added by _inject_dependent_update_ops.

Return type:

list[MigrateOperation | MigrateOp]

expand_update_ops(migration_ops)

Split Update*Op into Drop*Op + Create*Op.

All update op types are split so the topological sort can interleave drops and creates correctly:

• Views: CREATE OR REPLACE VIEW fails when a dependent view has an incompatible column list. Splitting lets dependents be dropped before the dependency is dropped and recreated.

• Functions/procedures: DROP FUNCTION fails when dependent triggers or views still reference the function. Splitting lets the sort order drops before the function drop and creates after the function create. _inject_dependent_update_ops ensures that dependent objects are present in the op list before this expansion runs.

• Triggers: CREATE OR REPLACE TRIGGER is not available in the PostgreSQL versions we target.

Return type:

list[MigrateOperation | MigrateOp]

prune_redundant_index_drops(ops)

Drop redundant DROP INDEX ops for tables dropped in this migration.

Postgres drops a table’s indexes when the table is dropped, so a separate DROP INDEX for a table that also has a DropTableOp in the same migration is redundant – and, when it lands after the table drop (e.g. wrapped in a ModifyTableOps that pins to the create phase), fails outright with “index does not exist”. Strip such index drops, whether top-level or nested inside a ModifyTableOps (drop the wrapper if it empties out).

Return type:

list[TypeVar(T, bound= MigrateOperation | MigrateOp)]

sort_migration_ops(migration_ops, *, fk_graph)

Return migration_ops topologically sorted by entity dependencies.

Dependency edges are derived from the ops themselves:

• A table depends on its schema.

• A table depends on tables it references via foreign keys.

• A replaceable entity (view, function, …) depends on its schema and on every schema-qualified table or view referenced in its SQL definition.

Only dependencies between ops in the current migration produce edges; references to already-existing objects are ignored.

Edge direction is determined per-op by phase: drop-phase ops reverse their edges (dependents dropped first), create-phase ops use normal direction (dependencies created first).

Parameters:

• migration_ops (list[MigrateOperation | MigrateOp]) – Operations from a single migration script.

• fk_graph (dict[tuple[str, str], set[tuple[str, str]]]) – FK dependency map (table -> referenced tables), built from build_fk_graph_from_metadata().

Return type:

list[MigrateOperation | MigrateOp]

Returns:

A new list containing the same ops in dependency order.

Automatic schema discovery for alembic autogenerate.

Scans a MetaData instance for schemas referenced by tables and views, then registers them with sqlalchemy-declarative-extensions so its built-in schema comparator emits the appropriate CREATE SCHEMA / DROP SCHEMA ops.

collect_schemas(metadata)

Return non-system schema names referenced by tables and views.

Return type:

set[str]

register_schemas(metadata)

Populate metadata.info["schemas"] from tables and views.

Merges with any schemas already registered on the metadata. Safe to call multiple times — existing entries are preserved.

Return type:

None

Custom alembic renderers that format SQL with pglast.

register_renderers()

Override the library’s renderers with pglast-formatted versions.

Return type:

None

render_item(type_, obj, autogen_context)

Render custom column types for migrations.

The in-scope types (codegen_database.types plus sqlalchemy_utils encrypted types, e.g. behind fsh_lib.oauth’s token columns) are TypeDecorator wrappers around plain SQL types, carrying Python-side state a migration can neither render nor use – an enum class, an encryption-key callable. Alembic’s default rendering emits their dotted class name with no import, producing a migration that doesn’t run. Migrations only need the database-level type, so this renders each type’s own impl (TextEnum -> sa.Text(), EncryptedText -> sa.Text(), …) instead of maintaining a parallel name -> DDL table.

Pass this as render_item to context.configure(...) in your Alembic env.py.

Returns False for unrecognized objects so Alembic falls through to its default rendering.

Return type:

str | Literal[False]

Pre-built features (codegen_database.ext.*)

Pre-built features composed from codegen_database primitives.

Each submodule under codegen_database.ext bundles tables, views, functions, plugins, and extensions that solve a specific problem (ledgers, audit queries, state machines, materialized-view refresh, cron, row-level security, etc.) using the declarative primitives exposed at the top level of codegen_database.

Import the specific submodule you need — this package does not re-export symbols.

Opt-in features composed from the primitives above. Each submodule is independent — import only the ones you use.

Ledger

Ledger configuration objects.

LedgerEvent configuration and helpers.

A LedgerEvent declares a named operation on a ledger. The user provides lambdas that produce SQLAlchemy selects; the plugin compiles them into a single PostgreSQL function per event.

Two modes are supported:

• Simple mode (input only): the input select is inserted directly into the ledger view.

• Diff mode (input + desired + existing): the desired state is diffed against the existing state and only the correcting deltas are inserted.

class LedgerEvent(name, input, desired=None, existing=None, diff_keys=<factory>)

Declare a named ledger operation.

Each event compiles into a single PostgreSQL function that inserts rows into the ledger view and returns the inserted rows via RETURNING *.

Simple mode — provide only input. The input select’s columns are inserted directly.

Diff mode — provide input, desired, existing, and diff_keys. The desired and existing selects are unioned and only non-zero deltas are inserted.

Parameters:

• name (str) – Unique event name within the ledger.

• input (Callable[[ParamCollector], Select]) – Lambda (p) -> Select that builds the input CTE. p is a ParamCollector.

• desired (Callable[[FromClause], SelectBase] | None) – Lambda (pginput) -> SelectBase that builds the desired-state CTE. pginput is a synthetic table reference to the input CTE – read function-param values as pginput.c.<name>, no p_ prefix. May return a union_all or other compound select.

• existing (Callable[[Table, FromClause, FromClause], Select] | None) – Lambda (table, desired, pginput) -> Select that builds the existing-state CTE. pginput is the same input-CTE reference passed to desired, so you can reach function params the same way – pginput.c.<name> – in the predicate or join. Use construct_ledger_balances_query() (filter by diff-key tuples in desired) or construct_ledger_scoped_balances_query() (filter by an owner predicate) for the common patterns.

• diff_keys (list[str]) – Column names used for grouping in diff mode. Required when desired is set.

class ParamCollector

Collect SQL function parameters during lambda evaluation.

Usage inside an input lambda:

lambda p: select(
    p("warehouse", String).label("warehouse"),
    p("sku", String).label("sku"),
)

Each call to p(name, sa_type) records the parameter and returns a literal_column() reference (p_name) suitable for embedding in a select.

Parameters:

None.

property function_params: list[FunctionParam]

Build the FunctionParam list for function registration.

Returns:

List of FunctionParam.input(...) entries.

construct_ledger_balances_query(*keys)

Return an existing callable for common balance lookup.

Produces a select that negates the current balances for each diff-key group present in the desired CTE:

SELECT key1, key2, SUM(value) * -1 AS value
FROM ledger_table
WHERE (key1, key2) IN (SELECT key1, key2 FROM desired)
GROUP BY key1, key2

Use this when every key the event might want to unwind is also present in desired. If a status transition or cancellation can retire keys – so they disappear from desired – use construct_ledger_scoped_balances_query() instead, which slices by an owner predicate rather than by the diff-key tuple.

Parameters:

*keys (str) – Dimension column names to group by.

Return type:

Callable[[Table, FromClause, FromClause], Select]

Returns:

A callable suitable for LedgerEvent(existing=...).

construct_ledger_scoped_balances_query(*keys, where)

Return an existing callable scoped by an owner predicate.

Unlike construct_ledger_balances_query(), which filters existing rows to diff-key tuples that appear in desired, this filters by an arbitrary predicate over the ledger root table:

SELECT key1, key2, SUM(value) * -1 AS value
FROM ledger_table
WHERE <where(root, pginput)>
GROUP BY key1, key2

The scoped form is the right tool whenever an event reconciles every ledger row belonging to some owner entity (a purchase order, an invoice, a work order) and a state transition might retire a diff key – i.e. a row that was posted previously should no longer be posted after the event runs. The default construct_ledger_balances_query would miss such retired keys because they no longer appear in desired; the scoped form sees them because it slices by the owner, so the diff naturally unwinds them.

Parameters:

• *keys (str) – Dimension column names to group by (the diff keys).

• where (Callable[[Table, FromClause], ColumnElement[bool]]) – Callable (ledger_root, pginput) -> predicate that returns a boolean restricting rows to the owner scope. pginput is a table reference to the generated input CTE, so function-param values are reached as pginput.c.<name> – the same accessor desired uses.

Return type:

Callable[[Table, FromClause, FromClause], Select]

Returns:

A callable suitable for LedgerEvent(existing=...).

Example

When the owner id is denormalized on the ledger:

def owned_by_input_po(ledger, pginput):
    return (
        ledger.c.purchase_order_id
        == pginput.c.purchase_order_id
    )

existing=construct_ledger_scoped_balances_query(
    "lot_id",
    "class_id",
    "account",
    "direction",
    where=owned_by_input_po,
)

When the owner is reached via a join, use a subquery:

def owned_by_input_po(ledger, pginput):
    return ledger.c.lot_id.in_(
        select(PurchaseLot.id)
        .join(LineItem, LineItem.id == PurchaseLot.line_item_id)
        .where(
            LineItem.purchase_order_id
            == pginput.c.purchase_order_id,
        ),
    )

existing=construct_ledger_scoped_balances_query(
    "lot_id",
    "class_id",
    "account",
    "direction",
    where=owned_by_input_po,
)

Query builders for ledger factories.

Provides pure-Python helpers that return SQLAlchemy selects without registering anything on metadata. Callers are responsible for passing the results to CodegenDatabasePlainView.

construct_ledger_balance_query(source, dimensions)

Build a balance aggregation query for a ledger.

Generates:

SELECT dim_col1, ..., SUM(value) AS balance
FROM <primary_table>
GROUP BY dim_col1, ...

Parameters:

• source (ContextSource) – A CodegenDatabaseLedger instance or a CodegenDatabaseBase subclass using CodegenDatabaseLedger as its factory.

• dimensions (list[str]) – Column names to group by. Must be a non-empty list.

Return type:

Select

Returns:

A SQLAlchemy Select.

Raises:

CodegenDatabaseValidationError – If dimensions is empty.

construct_ledger_latest_query(source, dimensions)

Build a latest-row query for a ledger using DISTINCT ON.

Generates:

SELECT * FROM <primary_table>
DISTINCT ON (dim_col1, ...)
ORDER BY dim_col1, ..., created_at DESC

Parameters:

• source (ContextSource) – A CodegenDatabaseLedger instance or a CodegenDatabaseBase subclass using CodegenDatabaseLedger as its factory.

• dimensions (list[str]) – Column names to partition by. Must be a non-empty list.

Return type:

Select

Returns:

A SQLAlchemy Select.

Raises:

CodegenDatabaseValidationError – If dimensions is empty.

Rolling and period aggregation query builders for ledger factories.

Provides pure-Python helpers that return SQLAlchemy selects without registering anything on metadata. Pass results to CodegenDatabasePlainView or CodegenDatabaseMaterializedView.

construct_ledger_gap_filled_period_rollup_query(source, dimensions, *, start=None, end=None, period='day')

Build a gap-filled period rollup query.

Generates the Cartesian product of every period in the range [start, end] with every dimension combination observed in the ledger, then LEFT JOINs deltas back in so empty periods are emitted with delta = 0:

SELECT p.period, d.dim_col1, ...,
       COALESCE(SUM(l.value), 0) AS delta
FROM generate_series(<start>, <end>,
                     INTERVAL '1 <period>') p(period)
CROSS JOIN (
    SELECT DISTINCT dim_col1, ...
    FROM <primary_table>
) d
LEFT JOIN <primary_table> l
  ON date_trunc('<period>', l.created_at) = p.period
  AND l.dim_col1 = d.dim_col1 AND ...
GROUP BY p.period, d.dim_col1, ...
ORDER BY p.period, d.dim_col1, ...

When start or end is None, a MIN(created_at) / MAX(created_at) subquery is substituted so the range auto-covers the observed data.

Parameters:

• source (ContextSource) – Ledger source.

• dimensions (list[str]) – Column names to group by. Must be a non-empty list.

• start (str | ColumnElement | None) – Lower bound of the period axis, or None to auto-detect. Accepts a timestamp literal string or a SQLAlchemy expression.

• end (str | ColumnElement | None) – Upper bound of the period axis, or None.

• period (Literal['microsecond', 'millisecond', 'second', 'minute', 'hour', 'day', 'week', 'month', 'quarter', 'year', 'decade', 'century', 'millennium']) – date_trunc precision string. Defaults to "day".

Return type:

Select

Returns:

A SQLAlchemy Select.

Raises:

CodegenDatabaseValidationError – If dimensions is empty or period is not a valid date_trunc precision.

construct_ledger_period_rollup_query(source, dimensions, period='day')

Build a period rollup query using date_trunc.

Generates:

SELECT date_trunc('day', created_at) AS period,
       dim_col1, ...,
       SUM(value) AS balance
FROM <primary_table>
GROUP BY 1, dim_col1, ...
ORDER BY 1, dim_col1, ...

Parameters:

• source (ContextSource) – A CodegenDatabaseLedger instance or a CodegenDatabaseBase subclass using CodegenDatabaseLedger as its factory.

• dimensions (list[str]) – Column names to group by. Must be a non-empty list.

• period (Literal['microsecond', 'millisecond', 'second', 'minute', 'hour', 'day', 'week', 'month', 'quarter', 'year', 'decade', 'century', 'millennium']) – date_trunc precision string, e.g. "day", "week", "month", "year". Defaults to "day".

Return type:

Select

Returns:

A SQLAlchemy Select.

Raises:

CodegenDatabaseValidationError – If dimensions is empty or period is not a valid date_trunc precision.

construct_ledger_period_running_balance_query(source, dimensions, period='day', split_by=None)

Build a per-period running balance query.

Generates:

SELECT date_trunc('<period>', created_at) AS period,
       dim_col1, ...,
       COALESCE(SUM(SUM(value)) OVER (
           PARTITION BY dim_col1, ...
           ORDER BY date_trunc('<period>', created_at)
           ROWS BETWEEN UNBOUNDED PRECEDING AND 1 PRECEDING
       ), 0) AS starting_balance,
       SUM(value) AS delta,
       SUM(SUM(value)) OVER (
           PARTITION BY dim_col1, ...
           ORDER BY date_trunc('<period>', created_at)
       ) AS ending_balance
FROM <primary_table>
GROUP BY 1, dim_col1, ...
ORDER BY 1, dim_col1, ...

Columns:

• starting_balance – cumulative balance just before this period (i.e. sum of all prior-period deltas). 0 for the first period in each partition.

• delta – net change within this period.

• ending_balance – cumulative balance at the close of this period (starting_balance + delta).

Passing split_by adds one SUM(value) FILTER (WHERE col = v) column per listed value, interleaved between delta and ending_balance. Typical use: split_by=("direction", ["debit", "credit"]) on a double-entry ledger.

Parameters:

• source (ContextSource) – A CodegenDatabaseLedger instance or a CodegenDatabaseBase subclass using CodegenDatabaseLedger as its factory.

• dimensions (list[str]) – Column names to partition by. Must be a non-empty list.

• period (Literal['microsecond', 'millisecond', 'second', 'minute', 'hour', 'day', 'week', 'month', 'quarter', 'year', 'decade', 'century', 'millennium']) – date_trunc precision string. Defaults to "day".

• split_by (tuple[str, list[str]] | None) – Optional (column_name, values) pair. Each value becomes a named SUM column filtered on equality.

Return type:

Select

Returns:

A SQLAlchemy Select.

Raises:

CodegenDatabaseValidationError – On empty dimensions, unknown period, empty or unknown split_by.

construct_ledger_pivoted_period_query(source, dimensions, periods, period='day')

Build a pivoted period query with one column per listed period.

Rows are dimension combinations; columns are the listed period start timestamps. Each cell holds the summed value for that period x dimension combination:

SELECT dim_col1, ...,
       COALESCE(SUM(value) FILTER (
           WHERE date_trunc('<period>', created_at)
               = '<periods[0]>'::timestamptz
       ), 0) AS "<periods[0]>",
       ...
FROM <primary_table>
GROUP BY dim_col1, ...
ORDER BY dim_col1, ...

Postgres requires the column set to be known at plan time, so periods must be passed in as an explicit list. This mirrors what the running-balance view provides but with periods on the column axis instead of rows, which is often the shape a spreadsheet or stacked bar chart wants.

Parameters:

• source (ContextSource) – Ledger source.

• dimensions (list[str]) – Column names to group by. Must be a non-empty list.

• periods (list[str]) – Period start timestamps as ISO strings, e.g. ["2024-01-01", "2024-01-02"]. Each becomes a named output column. Must be a non-empty list.

• period (Literal['microsecond', 'millisecond', 'second', 'minute', 'hour', 'day', 'week', 'month', 'quarter', 'year', 'decade', 'century', 'millennium']) – date_trunc precision string. Defaults to "day". Values in periods are matched after both sides are truncated to this precision.

Return type:

Select

Returns:

A SQLAlchemy Select.

Raises:

CodegenDatabaseValidationError – If dimensions or periods is empty, or period is invalid.

construct_ledger_rolling_window_query(source, dimensions, period='day', window_size=7)

Build a rolling-window sum query over per-period deltas.

Generates:

SELECT date_trunc('<period>', created_at) AS period,
       dim_col1, ...,
       SUM(value) AS delta,
       SUM(SUM(value)) OVER (
           PARTITION BY dim_col1, ...
           ORDER BY date_trunc('<period>', created_at)
           ROWS BETWEEN <window_size-1> PRECEDING
                        AND CURRENT ROW
       ) AS rolling_sum
FROM <primary_table>
GROUP BY 1, dim_col1, ...
ORDER BY 1, dim_col1, ...

The rolling sum is over rows, not calendar periods: missing periods are skipped rather than counted as zero. Use construct_ledger_gap_filled_period_rollup_query() as a CTE first if you need calendar-aligned rolling windows.

Parameters:

• source (ContextSource) – Ledger source.

• dimensions (list[str]) – Column names to partition by. Must be a non-empty list.

• period (Literal['microsecond', 'millisecond', 'second', 'minute', 'hour', 'day', 'week', 'month', 'quarter', 'year', 'decade', 'century', 'millennium']) – date_trunc precision string. Defaults to "day".

• window_size (int) – Number of rows in the trailing window, inclusive of the current row. Must be >= 1.

Return type:

Select

Returns:

A SQLAlchemy Select.

Raises:

CodegenDatabaseValidationError – If dimensions is empty, period is invalid, or window_size is less than 1.

construct_ledger_running_balance_query(source, dimensions)

Build a running balance query using a window function.

Generates:

SELECT dim_col1, ..., created_at, value,
       SUM(value) OVER (
           PARTITION BY dim_col1, ...
           ORDER BY created_at
           ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW
       ) AS running_balance
FROM <primary_table>
ORDER BY dim_col1, ..., created_at

Parameters:

• source (ContextSource) – A CodegenDatabaseLedger instance or a CodegenDatabaseBase subclass using CodegenDatabaseLedger as its factory.

• dimensions (list[str]) – Column names to partition by. Must be a non-empty list.

Return type:

Select

Returns:

A SQLAlchemy Select.

Raises:

CodegenDatabaseValidationError – If dimensions is empty.

Function spec builder for ledger factories.

Provides ledger_event_function() which compiles a LedgerEvent into a CodegenDatabaseFunctionSpec ready to unpack into CodegenDatabaseFunction.

ledger_event_function(source, event)

Build a function spec for a single ledger event.

Returns a CodegenDatabaseFunctionSpec.

The spec contains the SQL function body generated from the event lambdas and all metadata needed to register a PostgreSQL function via CodegenDatabaseFunction.

Parameters:

• source (ContextSource) – A CodegenDatabaseLedger instance or a CodegenDatabaseBase subclass using CodegenDatabaseLedger as its factory.

• event (LedgerEvent) – The LedgerEvent to compile into a function.

Return type:

CodegenDatabaseFunctionSpec

Returns:

A CodegenDatabaseFunctionSpec ready to pass to CodegenDatabaseFunction.

Raises:

CodegenDatabaseValidationError – If event configuration is invalid.

Chart-focused function builders for ledger factories.

These build CodegenDatabaseFunctionSpec instances that can be unpacked into a declarative CodegenDatabaseFunctionMixin subclass (via __funcspec__) or registered imperatively.

The generated functions are LANGUAGE sql STABLE: Postgres inlines them at plan time, so predicates added by callers (WHERE, LIMIT) push through the parameters. PL/pgSQL is avoided on purpose — dynamic grouping in PL/pgSQL means EXECUTE format(...) which blocks the planner from inlining.

Function bodies are assembled with SQLAlchemy core and compiled via compile_query() so identifiers, casts, and window frames go through the normal dialect compiler. Function parameters (p_period, p_start, p_end) surface as literal_column references so they render as bare identifiers in the generated SELECT.

Bucketing uses the codegen_database_date_bin helper rather than date_trunc so callers can pass arbitrary interval strides ('15 minutes', '3 months', '1 year', etc.). The helper must be installed in the same schema as the chart function (or in date_bin_schema, if overridden).

class NormalSide(*values)

Which side an account’s balance naturally grows on.

Double-entry accounting splits accounts into two camps:

• DEBIT: assets and expenses — the balance grows with debit postings (e.g. cash receipts, inventory purchases).

• CREDIT: liabilities, equity, and revenue — the balance grows with credit postings (e.g. sales, capital injections).

construct_double_entry_chart_function() reads a column carrying these string values from the accounts dimension and uses them to normalize delta = debits - credits (for debit-normal accounts) or delta = credits - debits (for credit-normal accounts), so delta is always positive when the balance moves in the account’s natural direction.

Pair with TextEnum to persist the string value:

from codegen_database import NormalSide, TextEnum

class Accounts(Base):
    name        = Column(String, unique=True, nullable=False)
    normal_side = Column(TextEnum(NormalSide), nullable=False)

construct_double_entry_chart_function(source, *, name, accounts_table, dimensions, account_column='account', direction_column='direction', accounts_key_column='name', normal_side_column='normal_side', period_default='1 day', date_bin_schema=None, date_bin_name='codegen_database_date_bin')

Build a double-entry period x dimensions chart function spec.

This is a specialization of construct_ledger_chart_function() for double-entry ledgers: it joins each raw row to an accounts dimension table, splits raw value into debits and credits by direction, and normalizes the per-bucket change (delta) by each account’s normal_side.

The normalization rule is the standard accounting convention:

• Debit-normal account (assets, expenses): delta = debits - credits.

• Credit-normal account (liabilities, equity, revenue): delta = credits - debits.

Running balances (starting_balance, ending_balance) are cumulative sums of the normalized delta inside the [p_start, p_end) window, partitioned by dimensions.

Signature:

<name>(p_period interval DEFAULT '<period_default>'::interval,
       p_start  timestamptz DEFAULT NULL,
       p_end    timestamptz DEFAULT NULL)
RETURNS TABLE (
    period tstzrange,
    <dim1> <type>, ...,
    starting_balance numeric,
    debits           numeric,
    credits          numeric,
    delta            numeric,
    ending_balance   numeric
)

The period column is the same [bucket_start, bucket_end) tstzrange as construct_ledger_chart_function() — see that function for filtering examples.

Body (CTE-based for readability; the CTE holds per-bucket aggregates, the outer SELECT adds running-balance windows):

WITH per_bucket AS (
    SELECT <bucket_start> AS bucket_start,
           <dims>,
           a.<normal_side> AS _normal_side,
           COALESCE(SUM(l.value) FILTER (
               WHERE l.<direction> = 'debit'), 0) AS debits,
           COALESCE(SUM(l.value) FILTER (
               WHERE l.<direction> = 'credit'), 0) AS credits
    FROM <schema>.<raw_table> l
    INNER JOIN <a_schema>.<accounts> a
        ON a.<name> = l.<account>
    WHERE (p_start IS NULL OR l.created_at >= p_start)
      AND (p_end   IS NULL OR l.created_at <  p_end)
    GROUP BY bucket_start, <dims>, a.<normal_side>
)
SELECT <period_range> AS period,
       <dims>,
       COALESCE(SUM(<delta>) OVER (
           PARTITION BY <dims>
           ORDER BY bucket_start
           ROWS BETWEEN UNBOUNDED PRECEDING AND 1 PRECEDING
       ), 0)::numeric AS starting_balance,
       debits::numeric AS debits,
       credits::numeric AS credits,
       (<delta>)::numeric AS delta,
       (SUM(<delta>) OVER (
           PARTITION BY <dims>
           ORDER BY bucket_start
       ))::numeric AS ending_balance
FROM per_bucket
ORDER BY bucket_start, <dims>

where <delta> is:

CASE WHEN _normal_side = 'debit'
     THEN debits - credits
     ELSE credits - debits
END

Note

starting_balance and ending_balance are cumulative within the filtered window, not absolute historical balances. Widen p_start to anchor against history.

Parameters:

• source (ContextSource) – Ledger source.

• name (str) – Unqualified function name.

• accounts_table (object) – A SQLAlchemy Table or declarative class for the accounts dimension. Must have name and normal_side columns (or override via accounts_key_column / normal_side_column).

• dimensions (list[str]) – Ledger columns to group by (typically includes account). Must be non-empty.

• account_column (str) – Name of the account column on the raw ledger table. Defaults to "account".

• direction_column (str) – Name of the direction column on the raw ledger table. Defaults to "direction".

• accounts_key_column (str) – Column on accounts_table that matches account_column values. Defaults to "name".

• normal_side_column (str) – Column on accounts_table holding 'debit' or 'credit'. Defaults to "normal_side".

• period_default (str) – Default for p_period as an interval literal, e.g. "1 day", "1 month".

• date_bin_schema (str | None) – Schema where codegen_database_date_bin lives. Defaults to the schema configured by ChartExtension ("codegen_database" unless overridden).

• date_bin_name (str) – Name of the polyfill function. Defaults to codegen_database_date_bin.

Return type:

CodegenDatabaseFunctionSpec

Returns:

A CodegenDatabaseFunctionSpec.

Raises:

CodegenDatabaseValidationError – If dimensions is empty, any dimension / account_column / direction_column is unknown, or period_default is empty.

construct_ledger_chart_function(source, *, name, dimensions, period_default='1 day', split_by=None, date_bin_schema=None, date_bin_name='codegen_database_date_bin')

Build a period x dimensions chart function spec.

This is a range function: callers pass p_start / p_end to bracket a half-open [p_start, p_end) window on created_at. Both default to NULL meaning “unbounded on that side”. The function returns one row per (codegen_database_date_bin(p_period, created_at), dimensions...) bucket inside the range.

Signature:

<name>(p_period interval DEFAULT '<period_default>'::interval,
       p_start  timestamptz DEFAULT NULL,
       p_end    timestamptz DEFAULT NULL)
RETURNS TABLE (
    period tstzrange,
    <dim1> <type>, ...,
    starting_balance numeric,
    delta  numeric,
    [<split_value_1> numeric, ...,]
    ending_balance numeric
)

The period column is a half-open [bucket_start, bucket_end) tstzrange so callers see the full extent of each bucket, not just its start. bucket_start is the polyfilled codegen_database_date_bin(p_period, created_at) and bucket_end is bucket_start + p_period. To filter:

-- All buckets contained in Feb 2024:
WHERE period <@ tstzrange('2024-02-01', '2024-03-01', '[)')

-- Exactly the Feb 2024 bucket (only useful when stride and
-- boundary align):
WHERE lower(period) = '2024-02-01'

Body (compiled from a SQLAlchemy select):

SELECT CASE
           WHEN <schema>.codegen_database_date_bin(p_period, created_at)
                IS NULL
               THEN NULL::tstzrange
           ELSE tstzrange(
               <schema>.codegen_database_date_bin(p_period, created_at),
               <schema>.codegen_database_date_bin(p_period, created_at)
                   + p_period,
               '[)')
       END AS period,
       <dims>,
       COALESCE(SUM(SUM(value)) OVER (
           PARTITION BY <dims>
           ORDER BY <schema>.codegen_database_date_bin(
               p_period, created_at)
           ROWS BETWEEN UNBOUNDED PRECEDING AND 1 PRECEDING
       ), 0)::numeric AS starting_balance,
       SUM(value)::numeric AS delta,
       [COALESCE(SUM(value) FILTER (
            WHERE <split_col> = '<v>'), 0)::numeric AS <v>, ...,]
       SUM(SUM(value)) OVER (
           PARTITION BY <dims>
           ORDER BY <schema>.codegen_database_date_bin(
               p_period, created_at)
       )::numeric AS ending_balance
FROM <schema>.<raw_table>
WHERE (p_start IS NULL OR created_at >= p_start)
  AND (p_end   IS NULL OR created_at <  p_end)
GROUP BY <schema>.codegen_database_date_bin(
    p_period, created_at), <dims>
ORDER BY <schema>.codegen_database_date_bin(
    p_period, created_at), <dims>

GROUP BY / ORDER BY use the scalar bucket-start rather than the range expression so the planner operates on timestamptz, not tstzrange, and window PARTITION BY stays cheap.

The codegen_database_date_bin helper (see construct_date_bin_function()) must exist in date_bin_schema. It polyfills Postgres’s date_bin for month/quarter/year strides, which the native function rejects. Mixed-stride intervals (e.g. '1 month 3 days') return NULL for period — the CASE in the body preserves that signal rather than emitting the unbounded range (,).

Note

starting_balance and ending_balance are cumulative flows within the filtered window, not absolute ledger balances. To anchor against history, widen p_start or compose with construct_ledger_balance_query().

Parameters:

• source (ContextSource) – Ledger source.

• name (str) – Unqualified function name.

• dimensions (list[str]) – Column names to group by. Must be non-empty.

• period_default (str) – Default value for p_period as a Postgres interval literal, e.g. "1 day", "15 minutes", "3 months", "1 year".

• split_by (tuple[str, list[str]] | None) – Optional (column_name, values) pair. Each value becomes a named SUM column filtered on equality, e.g. split_by=("direction", ["debit", "credit"]) on a double-entry ledger.

• date_bin_schema (str | None) – Schema where codegen_database_date_bin lives. Defaults to the schema configured by ChartExtension ("codegen_database" unless overridden).

• date_bin_name (str) – Name of the polyfill function. Defaults to codegen_database_date_bin.

Return type:

CodegenDatabaseFunctionSpec

Returns:

A CodegenDatabaseFunctionSpec.

Raises:

CodegenDatabaseValidationError – If dimensions is empty, any dimension is unknown, period_default is empty, or split_by names a missing column / is empty.

construct_ledger_pivoted_chart_function(source, *, name, dimensions, period_default='1 day', date_bin_schema=None, date_bin_name='codegen_database_date_bin')

Build a JSONB-pivoted chart function spec.

Returns one row per dimensions group with a single buckets jsonb column mapping UTC-normalized bucket-start keys to numeric deltas. Keeping the shape fixed (unlike crosstab, which forces callers to spell every pivot column in an AS t(...) clause at every call site) makes this function usable anywhere SELECT * FROM goes: ORMs, CTEs, views.

Signature:

<name>(p_period interval DEFAULT '<period_default>'::interval,
       p_start  timestamptz DEFAULT NULL,
       p_end    timestamptz DEFAULT NULL)
RETURNS TABLE (
    <dim1> <type>, ...,
    buckets jsonb
)

Typical usage:

SELECT warehouse, sku,
       (buckets->>'2024-01-01 00:00:00')::numeric AS day1,
       (buckets->>'2024-01-02 00:00:00')::numeric AS day2
FROM <schema>.<name>(
    p_period => '1 day'::interval,
    p_start  => '2024-01-01'::timestamptz,
    p_end    => '2024-01-03'::timestamptz
);

Bucket keys are the UTC-normalized text form of the bucket-start timestamp — to_char(bucket_start AT TIME ZONE 'UTC', 'YYYY-MM-DD HH24:MI:SS') — chosen over bucket_start::text so the keys are stable across session timezones. Missing buckets are absent from the JSONB (not present with value 0), so use COALESCE((buckets->>'...')::numeric, 0) when you need zeros.

Body (compiled from a SQLAlchemy select):

SELECT <dims>,
       jsonb_object_agg(
           to_char(bucket_start AT TIME ZONE 'UTC',
                   'YYYY-MM-DD HH24:MI:SS'),
           delta
       ) AS buckets
FROM (
    SELECT <dims>,
           <schema>.codegen_database_date_bin(p_period, created_at)
               AS bucket_start,
           SUM(value)::numeric AS delta
    FROM <fn_schema>.<raw_table>
    WHERE (p_start IS NULL OR created_at >= p_start)
      AND (p_end   IS NULL OR created_at <  p_end)
    GROUP BY <dims>, bucket_start
) AS b
GROUP BY <dims>
ORDER BY <dims>

Note

jsonb_object_agg raises on duplicate keys. The inner aggregate’s GROUP BY <dims>, bucket_start guarantees one row per key within each dimension group, so duplicates only appear if codegen_database_date_bin returns NULL (mixed-stride intervals like '1 month 3 days') for multiple source rows in the same dimension group — those keys collapse to null and Postgres rejects the aggregate. Avoid mixed-stride intervals; codegen_database_date_bin documents the supported forms.

Parameters:

• source (ContextSource) – Ledger source.

• name (str) – Unqualified function name.

• dimensions (list[str]) – Column names used as row keys. Must be non-empty.

• period_default (str) – Default for p_period as an interval literal, e.g. "1 day", "1 month".

• date_bin_schema (str | None) – Schema where codegen_database_date_bin lives. Defaults to the schema configured by ChartExtension.

• date_bin_name (str) – Name of the polyfill function. Defaults to codegen_database_date_bin.

Return type:

CodegenDatabaseFunctionSpec

Returns:

A CodegenDatabaseFunctionSpec.

Raises:

CodegenDatabaseValidationError – If dimensions is empty, any dimension is unknown, or period_default is empty.

construct_ledger_rollup_chart_function(source, *, name, dimensions, period_default='1 day', date_bin_schema=None, date_bin_name='codegen_database_date_bin')

Build a period x ROLLUP(dimensions) chart function spec.

This is a range function: callers pass p_start / p_end to bracket a half-open [p_start, p_end) window on created_at. Both default to NULL meaning “unbounded on that side”.

Signature:

<name>(p_period interval DEFAULT '<period_default>'::interval,
       p_start  timestamptz DEFAULT NULL,
       p_end    timestamptz DEFAULT NULL)
RETURNS TABLE (
    period tstzrange,
    <dim1> <type>, ...,
    delta  numeric
)

The period column is the same [bucket_start, bucket_end) tstzrange as construct_ledger_chart_function() — see that function for the rationale and for filtering examples. Unlike the regular chart function, starting_balance and ending_balance are not emitted: running totals are undefined against grouping sets.

The body uses ROLLUP to produce hierarchical subtotals in a single query. Rolled-up (subtotal) rows carry NULL in the rolled-up dimension columns; callers distinguish them from detail rows with <dim> IS NULL / <dim> IS NOT NULL. This assumes the dimension itself is never NULL in the underlying data — which is the norm for FK / enum dimensions; if your data has genuine NULL values in a dimension, subtotal rows will be indistinguishable from NULL-data rows.

SELECT CASE
           WHEN <schema>.codegen_database_date_bin(p_period, created_at)
                IS NULL
               THEN NULL::tstzrange
           ELSE tstzrange(
               <schema>.codegen_database_date_bin(p_period, created_at),
               <schema>.codegen_database_date_bin(p_period, created_at)
                   + p_period,
               '[)')
       END AS period,
       <dims>,
       SUM(value)::numeric AS delta
FROM <schema>.<raw_table>
WHERE (p_start IS NULL OR created_at >= p_start)
  AND (p_end   IS NULL OR created_at <  p_end)
GROUP BY <schema>.codegen_database_date_bin(p_period, created_at),
         ROLLUP(<dims>)
ORDER BY <schema>.codegen_database_date_bin(p_period, created_at),
         <dims> NULLS LAST

Parameters:

• source (ContextSource) – Ledger source.

• name (str) – Unqualified function name.

• dimensions (list[str]) – Column names forming the rollup hierarchy (left to right = outer to inner). Must be non-empty.

• period_default (str) – Default value for p_period as a Postgres interval literal, e.g. "1 day", "3 months".

• date_bin_schema (str | None) – Schema where codegen_database_date_bin lives. Defaults to the schema configured by ChartExtension ("codegen_database" unless overridden).

• date_bin_name (str) – Name of the polyfill function. Defaults to codegen_database_date_bin.

Return type:

CodegenDatabaseFunctionSpec

Returns:

A CodegenDatabaseFunctionSpec.

Raises:

CodegenDatabaseValidationError – If dimensions is empty, any dimension is unknown, or period_default is empty.

Chart extension for codegen_database ledgers.

ChartExtension wires the codegen_database_date_bin polyfill into the codegen_database lifecycle. Register it on a CodegenDatabaseConfig to make the ledger chart function builders usable:

from codegen_database.config import CodegenDatabaseConfig
from codegen_database.ext.chart import ChartExtension

config = CodegenDatabaseConfig()
config.use(ChartExtension())

After registration, codegen_database_date_bin lives in the codegen_database utility schema (CodegenDatabaseConfig.utility_schema, "codegen_database" by default). The ledger chart function builders find it via metadata.info["codegen_database_chart_schema"] and call into it with the correct schema qualification.

Without this extension, the chart function builders raise CodegenDatabaseValidationError with a message pointing the user at config.use(ChartExtension()).

class ChartExtension(name='codegen_database-chart', schema=None)

Register the codegen_database_date_bin polyfill into metadata.

When registered on a CodegenDatabaseConfig:

• configure_metadata creates codegen_database_date_bin in the resolved schema and stores it under metadata.info["codegen_database_chart_schema"] so the ledger chart function builders can resolve the helper without a user-supplied override.

• that schema is also added to metadata.info["schemas"] so Alembic autogenerate creates it.

Ledger chart function builders (construct_ledger_chart_function() and friends) call assert_chart_extension_declared() internally and raise CodegenDatabaseValidationError when this extension is absent.

Parameters:

• name (str) – Extension name. Defaults to "codegen_database-chart".

• schema (str | None) – Schema for the polyfill function. None (the default) uses the config’s utility_schema, so codegen_database_date_bin shares the one codegen_database utility schema with every other codegen_database-managed object. Set a string to override.

configure_metadata(metadata)

Register the polyfill, schema, and marker on metadata.

Return type:

None

assert_chart_extension_declared(metadata)

Raise if ChartExtension is not registered on metadata.

Called by the ledger chart function builders to give a clear error when the user forgot to register the extension on their CodegenDatabaseConfig.

Mirrors assert_pg_extension_declared(): if the marker is absent but a CodegenDatabaseConfig is present on metadata.info, the config’s extension hooks are run eagerly so the check can succeed.

Parameters:

metadata (MetaData) – The MetaData to check.

Raises:

CodegenDatabaseValidationError – If the extension is not declared and cannot be resolved from the registered config.

Return type:

None

codegen_database_date_bin polyfill: date_bin with month/year strides.

PostgreSQL’s built-in date_bin(stride interval, ts timestamptz, origin timestamptz) refuses strides that contain units of month or larger. The reason is implementation-driven — date_bin works by subtracting ts and origin and flooring that interval by stride, which requires uniform-length strides, and months are 28-31 days.

This module registers an SQL function codegen_database_date_bin that bridges the gap by falling through to date_bin for sub-month strides and switching to calendar arithmetic for whole-month strides (including quarter and year). Mixed strides (e.g. '1 month 3 days') are rejected by returning NULL.

The helper is used by the ledger chart functions in codegen_database.ext.ledger.chart_functions, which accept an interval p_period parameter. It lives under codegen_database.ext.chart rather than the ledger because ChartExtension is what installs it during configure_metadata; the ledger merely calls into it.

CODEGEN_DATABASE_DATE_BIN_DEFAULT_ORIGIN = "'2000-01-03'::timestamptz"

Default origin used by the polyfill (a Monday at midnight local).

'2000-01-03' was chosen so that week strides align to Monday (ISO-style), day/hour strides align to midnight, and month/quarter/ year strides align to the 1st of the month (because the month path snaps via date_trunc('month', origin)).

CODEGEN_DATABASE_DATE_BIN_NAME = 'codegen_database_date_bin'

Canonical unqualified name of the polyfill function.

construct_date_bin_function(*, name='codegen_database_date_bin')

Build a codegen_database_date_bin function spec.

Returns a CodegenDatabaseFunctionSpec. Register the returned spec once per schema that needs it (typically wherever a ledger chart function lives). The ledger chart functions call <schema>.codegen_database_date_bin(p_period, created_at) with the chart function’s own schema by default.

Usage:

class DateBin(CodegenDatabaseFunctionMixin, Base):
    __table_args__ = {"schema": "ops"}
    __funcspec__ = construct_date_bin_function()

Signature:

codegen_database_date_bin(
    stride interval,
    ts     timestamptz,
    origin timestamptz DEFAULT '2000-01-03'::timestamptz
) RETURNS timestamptz

Semantics:

• stride contains only sub-month units → delegate to Postgres’s built-in date_bin.

• stride is a whole number of months (so includes quarter and year) → calendar arithmetic, aligned to the 1st of the month.

• stride mixes month and sub-month units → returns NULL (no uniform bucketing possible).

Parameters:

name (str) – Unqualified function name. Defaults to "codegen_database_date_bin". Chart functions will look for the helper under this name, so override only if you know what you are doing.

Return type:

CodegenDatabaseFunctionSpec

Returns:

A CodegenDatabaseFunctionSpec ready to pass to CodegenDatabaseFunction or assign to __funcspec__ on a CodegenDatabaseFunctionMixin subclass.

Audit

Audit/history query helpers.

These helpers return SQLAlchemy Selects without registering anything on metadata. Pass the result to CodegenDatabasePlainView (or CodegenDatabaseMaterializedView) to expose it as a view, or execute it directly against a connection.

Two dimension layouts are supported, each with its own pair of helpers:

• Append-only (SCD Type 2) – construct_append_only_diff_query(), construct_append_only_changed_between_query(). Each row is a full version. For CodegenDatabaseAppendOnly factories, pass factory.ctx["attributes"].

• EAV – construct_eav_diff_query(), construct_eav_changed_between_query(). Each row is one attribute’s value at a point in time. For CodegenDatabaseEAV factories, pass factory.ctx["attribute"].

class ActivityViewPlugin

Generate a {table_name}_activity changeset view.

Auto-included by both append-only and EAV factories. Detects which backing table is present (attributes for append-only, attribute for EAV) and builds the activity view accordingly.

The view is keyed by the backing table’s BigInteger surrogate id so keyset pagination works – the entity UUID repeats across versions and can’t page.

run(ctx)

Create the activity view on the shared metadata.

Return type:

None

construct_append_only_activity_query(table, *, key_cols, tracked_cols=None, ts_col='created_at')

Unified activity stream over an append-only (SCD Type 2) table.

Wraps the same LAG() partitioned by key_cols that construct_append_only_diff_query() uses, then unpivots the wide <col>_before / <col>_after pairs into one row per changed field – so a transition that touched three columns yields three activity rows (one per field) rather than one wide row. Each row carries actor = NULL (Path A) and a change_type of "create" (first version of the key, before IS NULL) or "update".

Generates (conceptually):

WITH lagged AS (
  SELECT <key...>, <ts> AS changed_at,
         lag(<col1>) OVER w AS <col1>_before, <col1> AS <col1>_after,
         ...
  FROM <table>
  WINDOW w AS (PARTITION BY <key...> ORDER BY <ts>)
)
SELECT <key...>, changed_at, NULL AS actor, '<col1>' AS field,
       <col1>_before::text AS before, <col1>_after::text AS after,
       CASE WHEN <col1>_before IS NULL THEN 'create' ELSE 'update' END
         AS change_type
FROM lagged WHERE <col1>_before IS DISTINCT FROM <col1>_after
UNION ALL
SELECT ... '<col2>' ...
WHERE <col2>_before IS DISTINCT FROM <col2>_after
...
ORDER BY <key...>, changed_at, field

Parameters:

• table (FromClause) – The append-only history table.

• key_cols (list[str]) – Column names identifying the audited entity (the LAG partition).

• tracked_cols (list[str] | None) – Columns whose transitions to surface. Defaults to every non-key, non-timestamp, non-PK column.

• ts_col (str) – Timestamp column used for ordering. Defaults to "created_at".

Return type:

CompoundSelect

Returns:

A SQLAlchemy Select with columns (*key_cols, changed_at, actor, field, before, after, change_type).

Raises:

CodegenDatabaseValidationError – If key_cols is empty or a referenced column is missing.

construct_append_only_changed_between_query(table, *, key_cols, start, end, tracked_cols=None, ts_col='created_at')

Net-change query over a window on an append-only history table.

Companion to construct_append_only_diff_query(). For each key, compares state-as-of start against state-as-of end (both inclusive). Intermediate versions inside the window are collapsed away – only the net delta is returned. One row per key that changed (including keys created or last-seen inside the window, which appear with NULL on the missing side). See construct_eav_changed_between_query() for the EAV equivalent.

Generates (conceptually):

WITH
  before_ AS (
    SELECT DISTINCT ON (<key...>) <key...>, <col...>
    FROM <table> WHERE <ts> <= :start
    ORDER BY <key...>, <ts> DESC
  ),
  after_ AS (
    SELECT DISTINCT ON (<key...>) <key...>, <col...>
    FROM <table> WHERE <ts> <= :end
    ORDER BY <key...>, <ts> DESC
  )
SELECT COALESCE(a.<key>, b.<key>) AS <key>, ...,
       b.<col> AS <col>_before, a.<col> AS <col>_after, ...
FROM after_ a FULL JOIN before_ b USING (<key...>)
WHERE b.<col> IS DISTINCT FROM a.<col> OR ...

Parameters:

• table (FromClause) – The append-only history table.

• key_cols (list[str]) – Column names identifying the audited entity.

• start (datetime | ColumnElement[Any]) – Lower bound of the window (inclusive). A datetime or SQL expression.

• end (datetime | ColumnElement[Any]) – Upper bound of the window (inclusive).

• tracked_cols (list[str] | None) – Columns whose net change to surface. Defaults to every non-key, non-timestamp, non-PK column.

• ts_col (str) – Timestamp column. Defaults to "created_at".

Return type:

Select

Returns:

A SQLAlchemy Select with columns (*key_cols, <col>_before, <col>_after, ...).

Raises:

CodegenDatabaseValidationError – If key_cols is empty, a referenced column is missing, or tracked_cols resolves to empty.

construct_append_only_diff_query(table, *, key_cols, tracked_cols=None, ts_col='created_at')

Per-transition diff over an append-only (SCD Type 2) history table.

Works on any table where each row represents a full version of an entity – typically the attributes backing table of a CodegenDatabaseAppendOnly factory (factory.ctx["attributes"]), or a hand-rolled append-only log. See construct_eav_diff_query() for the EAV equivalent.

Uses LAG() partitioned by key_cols ordered by ts_col to pair each row with its predecessor. Emits one row per transition where at least one tracked column changed (IS DISTINCT FROM).

The first row per key is included as a creation event: its *_before columns are NULL and *_after hold the initial values.

Generates (conceptually):

SELECT <key...>,
       <ts>                                      AS changed_at,
       LAG(<col1>) OVER w AS <col1>_before, <col1> AS <col1>_after,
       LAG(<col2>) OVER w AS <col2>_before, <col2> AS <col2>_after,
       ...
FROM <table>
WINDOW w AS (PARTITION BY <key...> ORDER BY <ts>)
WHERE <col1>_before IS DISTINCT FROM <col1>_after
   OR <col2>_before IS DISTINCT FROM <col2>_after
   OR ...
ORDER BY <key...>, changed_at

Parameters:

• table (FromClause) – The append-only history table.

• key_cols (list[str]) – Column names identifying the audited entity (the partition for LAG). Must be non-empty.

• tracked_cols (list[str] | None) – Columns whose transitions to surface. Defaults to every column that is not in key_cols, not the timestamp, and not part of the primary key.

• ts_col (str) – Timestamp column used for ordering. Defaults to "created_at".

Return type:

Select

Returns:

A SQLAlchemy Select with columns (*key_cols, changed_at, <col>_before, <col>_after, ...).

Raises:

CodegenDatabaseValidationError – If key_cols is empty, a referenced column is missing, or tracked_cols resolves to empty.

construct_eav_activity_query(table, *, entity_col='entity_id', attribute_col='attribute_name', value_cols=None, attributes=None, ts_col='created_at')

Unified activity stream over an EAV attribute log.

The EAV diff is already long (one row per (entity, attribute) transition), so this wraps construct_eav_diff_query() and adds the two missing unified columns: actor = NULL (Path A) and change_type ("create" for the first observation of an attribute, "update" otherwise).

Parameters:

• table (FromClause) – The EAV attribute log.

• entity_col (str) – Column naming the entity. Defaults to "entity_id".

• attribute_col (str) – Column naming the attribute. Defaults to "attribute_name".

• value_cols (list[str] | None) – The typed value columns. Defaults to every column ending in "_value".

• attributes (list[str] | None) – Optional list of attribute names to restrict to.

• ts_col (str) – Timestamp column. Defaults to "created_at".

Return type:

Select

Returns:

A SQLAlchemy Select with columns (<entity_col>, <attribute_col>, changed_at, actor, field, before, after, change_type). field mirrors the attribute name so the column layout matches the append-only helper.

Raises:

CodegenDatabaseValidationError – If a referenced column is missing.

construct_eav_changed_between_query(table, *, start, end, entity_col='entity_id', attribute_col='attribute_name', value_cols=None, attributes=None, ts_col='created_at')

Net-change query over a window on an EAV attribute log.

Companion to construct_eav_diff_query(). For each (entity, attribute) pair, compares value-as-of start against value-as-of end (both inclusive) via two DISTINCT ON snapshots joined with a FULL JOIN. One row per pair whose value differs; pairs introduced inside the window appear with value_before = NULL.

Parameters:

• table (FromClause) – The EAV attribute log.

• start (datetime | ColumnElement[Any]) – Lower bound of the window (inclusive).

• end (datetime | ColumnElement[Any]) – Upper bound of the window (inclusive).

• entity_col (str) – Column naming the entity.

• attribute_col (str) – Column naming the attribute.

• value_cols (list[str] | None) – Typed value columns; defaults to *_value.

• attributes (list[str] | None) – Optional attribute-name filter.

• ts_col (str) – Timestamp column.

Return type:

Select

Returns:

A SQLAlchemy Select with columns (<entity_col>, <attribute_col>, value_before, value_after).

Raises:

CodegenDatabaseValidationError – If a referenced column is missing or value_cols resolves to empty.

construct_eav_diff_query(table, *, entity_col='entity_id', attribute_col='attribute_name', value_cols=None, attributes=None, ts_col='created_at')

Per-attribute diff over an EAV attribute log.

Works on any table shaped like the codegen_database EAV attribute table: one row per (entity, attribute) value observation. For a CodegenDatabaseEAV factory, pass factory.ctx["attribute"]. See construct_append_only_diff_query() for the append-only equivalent.

Uses LAG() partitioned by (entity, attribute) ordered by ts_col. The first observation per (entity, attribute) comes through with value_before = NULL (a “first seen” event).

All value columns are coalesced into a single text-typed expression. EAV rows carry exactly one non-null value, so the coalesce is loss-free modulo formatting.

Generates (conceptually):

SELECT entity, attribute, <ts> AS changed_at,
       LAG(value) OVER w AS value_before,
       value        AS value_after
FROM (
    SELECT <entity>, <attribute>, <ts>,
           COALESCE(<v1>::text, <v2>::text, ...) AS value
    FROM <table>
    [WHERE <attribute> IN :attributes]
) t
WINDOW w AS (PARTITION BY entity, attribute ORDER BY <ts>)
WHERE value_before IS DISTINCT FROM value_after
ORDER BY entity, attribute, changed_at

Parameters:

• table (FromClause) – The EAV attribute log.

• entity_col (str) – Column naming the entity. Defaults to "entity_id".

• attribute_col (str) – Column naming the attribute. Defaults to "attribute_name".

• value_cols (list[str] | None) – The typed value columns. Defaults to every column ending in "_value".

• attributes (list[str] | None) – Optional list of attribute names to restrict the diff to. None means all attributes.

• ts_col (str) – Timestamp column. Defaults to "created_at".

Return type:

Select

Returns:

A SQLAlchemy Select with columns (<entity_col>, <attribute_col>, changed_at, value_before, value_after).

Raises:

CodegenDatabaseValidationError – If a referenced column is missing or value_cols resolves to empty.

Query builders for history/audit tables.

Pure-Python helpers that return SQLAlchemy selects. Two dimension layouts are supported:

• Append-only (SCD Type 2): each version is a full row, keyed by the dimension’s business columns. Use construct_append_only_diff_query() and construct_append_only_changed_between_query(). For a CodegenDatabaseAppendOnly factory, pass factory.ctx["attributes"].

• EAV (Entity-Attribute-Value): each row is a single attribute’s value at a point in time. Use construct_eav_diff_query() and construct_eav_changed_between_query(). For a CodegenDatabaseEAV factory, pass factory.ctx["attribute"].

All helpers emit typed <col>_before / <col>_after (or value_before / value_after for EAV) pairs. No JSONB.

construct_append_only_changed_between_query(table, *, key_cols, start, end, tracked_cols=None, ts_col='created_at')

Net-change query over a window on an append-only history table.

Companion to construct_append_only_diff_query(). For each key, compares state-as-of start against state-as-of end (both inclusive). Intermediate versions inside the window are collapsed away – only the net delta is returned. One row per key that changed (including keys created or last-seen inside the window, which appear with NULL on the missing side). See construct_eav_changed_between_query() for the EAV equivalent.

Generates (conceptually):

WITH
  before_ AS (
    SELECT DISTINCT ON (<key...>) <key...>, <col...>
    FROM <table> WHERE <ts> <= :start
    ORDER BY <key...>, <ts> DESC
  ),
  after_ AS (
    SELECT DISTINCT ON (<key...>) <key...>, <col...>
    FROM <table> WHERE <ts> <= :end
    ORDER BY <key...>, <ts> DESC
  )
SELECT COALESCE(a.<key>, b.<key>) AS <key>, ...,
       b.<col> AS <col>_before, a.<col> AS <col>_after, ...
FROM after_ a FULL JOIN before_ b USING (<key...>)
WHERE b.<col> IS DISTINCT FROM a.<col> OR ...

Parameters:

• table (FromClause) – The append-only history table.

• key_cols (list[str]) – Column names identifying the audited entity.

• start (datetime | ColumnElement[Any]) – Lower bound of the window (inclusive). A datetime or SQL expression.

• end (datetime | ColumnElement[Any]) – Upper bound of the window (inclusive).

• tracked_cols (list[str] | None) – Columns whose net change to surface. Defaults to every non-key, non-timestamp, non-PK column.

• ts_col (str) – Timestamp column. Defaults to "created_at".

Return type:

Select

Returns:

A SQLAlchemy Select with columns (*key_cols, <col>_before, <col>_after, ...).

Raises:

CodegenDatabaseValidationError – If key_cols is empty, a referenced column is missing, or tracked_cols resolves to empty.

construct_append_only_diff_query(table, *, key_cols, tracked_cols=None, ts_col='created_at')

Per-transition diff over an append-only (SCD Type 2) history table.

Works on any table where each row represents a full version of an entity – typically the attributes backing table of a CodegenDatabaseAppendOnly factory (factory.ctx["attributes"]), or a hand-rolled append-only log. See construct_eav_diff_query() for the EAV equivalent.

Uses LAG() partitioned by key_cols ordered by ts_col to pair each row with its predecessor. Emits one row per transition where at least one tracked column changed (IS DISTINCT FROM).

The first row per key is included as a creation event: its *_before columns are NULL and *_after hold the initial values.

Generates (conceptually):

SELECT <key...>,
       <ts>                                      AS changed_at,
       LAG(<col1>) OVER w AS <col1>_before, <col1> AS <col1>_after,
       LAG(<col2>) OVER w AS <col2>_before, <col2> AS <col2>_after,
       ...
FROM <table>
WINDOW w AS (PARTITION BY <key...> ORDER BY <ts>)
WHERE <col1>_before IS DISTINCT FROM <col1>_after
   OR <col2>_before IS DISTINCT FROM <col2>_after
   OR ...
ORDER BY <key...>, changed_at

Parameters:

• table (FromClause) – The append-only history table.

• key_cols (list[str]) – Column names identifying the audited entity (the partition for LAG). Must be non-empty.

• tracked_cols (list[str] | None) – Columns whose transitions to surface. Defaults to every column that is not in key_cols, not the timestamp, and not part of the primary key.

• ts_col (str) – Timestamp column used for ordering. Defaults to "created_at".

Return type:

Select

Returns:

A SQLAlchemy Select with columns (*key_cols, changed_at, <col>_before, <col>_after, ...).

Raises:

CodegenDatabaseValidationError – If key_cols is empty, a referenced column is missing, or tracked_cols resolves to empty.

construct_eav_changed_between_query(table, *, start, end, entity_col='entity_id', attribute_col='attribute_name', value_cols=None, attributes=None, ts_col='created_at')

Net-change query over a window on an EAV attribute log.

Companion to construct_eav_diff_query(). For each (entity, attribute) pair, compares value-as-of start against value-as-of end (both inclusive) via two DISTINCT ON snapshots joined with a FULL JOIN. One row per pair whose value differs; pairs introduced inside the window appear with value_before = NULL.

Parameters:

• table (FromClause) – The EAV attribute log.

• start (datetime | ColumnElement[Any]) – Lower bound of the window (inclusive).

• end (datetime | ColumnElement[Any]) – Upper bound of the window (inclusive).

• entity_col (str) – Column naming the entity.

• attribute_col (str) – Column naming the attribute.

• value_cols (list[str] | None) – Typed value columns; defaults to *_value.

• attributes (list[str] | None) – Optional attribute-name filter.

• ts_col (str) – Timestamp column.

Return type:

Select

Returns:

A SQLAlchemy Select with columns (<entity_col>, <attribute_col>, value_before, value_after).

Raises:

CodegenDatabaseValidationError – If a referenced column is missing or value_cols resolves to empty.

construct_eav_diff_query(table, *, entity_col='entity_id', attribute_col='attribute_name', value_cols=None, attributes=None, ts_col='created_at')

Per-attribute diff over an EAV attribute log.

Works on any table shaped like the codegen_database EAV attribute table: one row per (entity, attribute) value observation. For a CodegenDatabaseEAV factory, pass factory.ctx["attribute"]. See construct_append_only_diff_query() for the append-only equivalent.

Uses LAG() partitioned by (entity, attribute) ordered by ts_col. The first observation per (entity, attribute) comes through with value_before = NULL (a “first seen” event).

All value columns are coalesced into a single text-typed expression. EAV rows carry exactly one non-null value, so the coalesce is loss-free modulo formatting.

Generates (conceptually):

SELECT entity, attribute, <ts> AS changed_at,
       LAG(value) OVER w AS value_before,
       value        AS value_after
FROM (
    SELECT <entity>, <attribute>, <ts>,
           COALESCE(<v1>::text, <v2>::text, ...) AS value
    FROM <table>
    [WHERE <attribute> IN :attributes]
) t
WINDOW w AS (PARTITION BY entity, attribute ORDER BY <ts>)
WHERE value_before IS DISTINCT FROM value_after
ORDER BY entity, attribute, changed_at

Parameters:

• table (FromClause) – The EAV attribute log.

• entity_col (str) – Column naming the entity. Defaults to "entity_id".

• attribute_col (str) – Column naming the attribute. Defaults to "attribute_name".

• value_cols (list[str] | None) – The typed value columns. Defaults to every column ending in "_value".

• attributes (list[str] | None) – Optional list of attribute names to restrict the diff to. None means all attributes.

• ts_col (str) – Timestamp column. Defaults to "created_at".

Return type:

Select

Returns:

A SQLAlchemy Select with columns (<entity_col>, <attribute_col>, changed_at, value_before, value_after).

Raises:

CodegenDatabaseValidationError – If a referenced column is missing or value_cols resolves to empty.

Unified activity/audit query helpers.

Companion to codegen_database.ext.audit.queries. Where the construct_*_diff_query helpers emit the raw per-table transition shape (wide <col>_before / <col>_after pairs for append-only, long value_before / value_after for EAV), these helpers fold both layouts into one uniform activity row:

(key..., changed_at, actor, field, before, after, change_type)

• changed_at – when the transition happened.

• actor – who made it. NULL for historical rows: today’s append-only / EAV backing tables carry only created_at, no actor column (Path A – non-invasive). Forward-only capture plugs in here once a resource records an actor.

• field – the column (append-only) or attribute name (EAV) that changed.

• before / after – the prior and new values, cast to text for a uniform wire shape (the FE renders them as strings).

• change_type – "create" for the first observation of a key / attribute (before IS NULL), "update" otherwise.

Both helpers return a SQLAlchemy Select without registering anything on metadata. Execute directly against a connection.

construct_append_only_activity_query(table, *, key_cols, tracked_cols=None, ts_col='created_at')

Unified activity stream over an append-only (SCD Type 2) table.

Wraps the same LAG() partitioned by key_cols that construct_append_only_diff_query() uses, then unpivots the wide <col>_before / <col>_after pairs into one row per changed field – so a transition that touched three columns yields three activity rows (one per field) rather than one wide row. Each row carries actor = NULL (Path A) and a change_type of "create" (first version of the key, before IS NULL) or "update".

Generates (conceptually):

WITH lagged AS (
  SELECT <key...>, <ts> AS changed_at,
         lag(<col1>) OVER w AS <col1>_before, <col1> AS <col1>_after,
         ...
  FROM <table>
  WINDOW w AS (PARTITION BY <key...> ORDER BY <ts>)
)
SELECT <key...>, changed_at, NULL AS actor, '<col1>' AS field,
       <col1>_before::text AS before, <col1>_after::text AS after,
       CASE WHEN <col1>_before IS NULL THEN 'create' ELSE 'update' END
         AS change_type
FROM lagged WHERE <col1>_before IS DISTINCT FROM <col1>_after
UNION ALL
SELECT ... '<col2>' ...
WHERE <col2>_before IS DISTINCT FROM <col2>_after
...
ORDER BY <key...>, changed_at, field

Parameters:

• table (FromClause) – The append-only history table.

• key_cols (list[str]) – Column names identifying the audited entity (the LAG partition).

• tracked_cols (list[str] | None) – Columns whose transitions to surface. Defaults to every non-key, non-timestamp, non-PK column.

• ts_col (str) – Timestamp column used for ordering. Defaults to "created_at".

Return type:

CompoundSelect

Returns:

A SQLAlchemy Select with columns (*key_cols, changed_at, actor, field, before, after, change_type).

Raises:

CodegenDatabaseValidationError – If key_cols is empty or a referenced column is missing.

construct_eav_activity_query(table, *, entity_col='entity_id', attribute_col='attribute_name', value_cols=None, attributes=None, ts_col='created_at')

Unified activity stream over an EAV attribute log.

The EAV diff is already long (one row per (entity, attribute) transition), so this wraps construct_eav_diff_query() and adds the two missing unified columns: actor = NULL (Path A) and change_type ("create" for the first observation of an attribute, "update" otherwise).

Parameters:

• table (FromClause) – The EAV attribute log.

• entity_col (str) – Column naming the entity. Defaults to "entity_id".

• attribute_col (str) – Column naming the attribute. Defaults to "attribute_name".

• value_cols (list[str] | None) – The typed value columns. Defaults to every column ending in "_value".

• attributes (list[str] | None) – Optional list of attribute names to restrict to.

• ts_col (str) – Timestamp column. Defaults to "created_at".

Return type:

Select

Returns:

A SQLAlchemy Select with columns (<entity_col>, <attribute_col>, changed_at, actor, field, before, after, change_type). field mirrors the attribute name so the column layout matches the append-only helper.

Raises:

CodegenDatabaseValidationError – If a referenced column is missing.

Comments

Comment query helpers.

Pure-Python helpers that return SQLAlchemy selects. One helper is exposed:

• construct_comments_thread_query() – the flat, parent-first row stream for one (resource, row), ready to feed fsh_lib.comments.build_comment_thread().

Pair with fsh_lib.comments.CommentMixin for the storage columns.

construct_comments_thread_query(table, *, resource_type, resource_id, resource_type_col='resource_type', resource_id_col='resource_id', parent_col='parent_comment_id', created_at_col='created_at')

Flat, parent-first comment stream for one (resource, row).

Filters the comments table to resource_type / resource_id and orders so every parent precedes its children: top-level comments (parent_comment_id IS NULL) first by created_at, then each parent’s replies by created_at. Feed the result to fsh_lib.comments.build_comment_thread() to nest it.

The ordering groups top-level comments (parent_comment_id IS NULL) ahead of replies, and within each group sorts by created_at. fsh_lib.comments.build_comment_thread() nests the flat stream into a tree from an id-keyed dict, so the only ordering that matters for the tree is the order of a parent’s replies – which this sort keeps chronological. This avoids a recursive CTE for the common one-level-deep thread shape.

Generates (conceptually):

SELECT *
FROM <table>
WHERE resource_type = :resource_type
  AND resource_id   = :resource_id
ORDER BY parent_comment_id NULLS FIRST, created_at

Parameters:

• table (Table) – The comments table.

• resource_type (str) – Slug of the parent resource to filter on.

• resource_id (str) – Stringified id of the parent row to filter on.

• resource_type_col (str) – Column holding the resource slug.

• resource_id_col (str) – Column holding the parent row’s id.

• parent_col (str) – Self-referential parent-comment column.

• created_at_col (str) – Timestamp column for child ordering.

Return type:

Select

Returns:

A SQLAlchemy Select over every column of table, filtered and ordered for threading.

Raises:

CodegenDatabaseValidationError – If a referenced column is missing.

Threaded comment query helpers.

These helpers return SQLAlchemy Selects without registering anything on metadata. Execute the result directly against a connection. Pair with fsh_lib.comments.CommentMixin (the storage columns) and fsh_lib.comments.build_comment_thread() (the pure-Python nester) to serve a threaded comment list for any opted-in resource.

construct_comments_thread_query() returns the flat, parent-first row stream fsh_lib.comments.build_comment_thread() nests into a tree. The ordering is structural (every parent precedes its children) so the nester never has to look ahead.

construct_comments_thread_query(table, *, resource_type, resource_id, resource_type_col='resource_type', resource_id_col='resource_id', parent_col='parent_comment_id', created_at_col='created_at')

Flat, parent-first comment stream for one (resource, row).

Filters the comments table to resource_type / resource_id and orders so every parent precedes its children: top-level comments (parent_comment_id IS NULL) first by created_at, then each parent’s replies by created_at. Feed the result to fsh_lib.comments.build_comment_thread() to nest it.

The ordering groups top-level comments (parent_comment_id IS NULL) ahead of replies, and within each group sorts by created_at. fsh_lib.comments.build_comment_thread() nests the flat stream into a tree from an id-keyed dict, so the only ordering that matters for the tree is the order of a parent’s replies – which this sort keeps chronological. This avoids a recursive CTE for the common one-level-deep thread shape.

Generates (conceptually):

SELECT *
FROM <table>
WHERE resource_type = :resource_type
  AND resource_id   = :resource_id
ORDER BY parent_comment_id NULLS FIRST, created_at

Parameters:

• table (Table) – The comments table.

• resource_type (str) – Slug of the parent resource to filter on.

• resource_id (str) – Stringified id of the parent row to filter on.

• resource_type_col (str) – Column holding the resource slug.

• resource_id_col (str) – Column holding the parent row’s id.

• parent_col (str) – Self-referential parent-comment column.

• created_at_col (str) – Timestamp column for child ordering.

Return type:

Select

Returns:

A SQLAlchemy Select over every column of table, filtered and ordered for threading.

Raises:

CodegenDatabaseValidationError – If a referenced column is missing.

State machines

Ledger-backed state machine factory.

class CodegenDatabaseStateMachine(name, schemaname, metadata, subject_column, states, valid_transitions, extra_columns=None, *, actor_column=None, actor_fk=None, allow_initial=True)

Ledger-backed state machine with transition validation.

Creates an append-only transitions table, a full history view, and a current-state view. A BEFORE INSERT trigger validates that each new transition is:

1. Declared in valid_transitions.

2. Consistent with the subject’s actual current state.

The valid transitions are compiled into the trigger function body at migration time, so changing them requires a migration.

Parameters:

• name (str) – Base name for all generated objects.

• schemaname (str) – PostgreSQL schema for all objects.

• metadata (MetaData) – SQLAlchemy MetaData to register on.

• subject_column (Column) – Column that identifies the entity being tracked. Use CodegenDatabaseForeignKey to reference the subject’s table. Must be nullable=False.

• states (list[str]) – List of valid state strings.

• valid_transitions (list[tuple[str | None, str]]) – List of (from_state, to_state) pairs. Use None as from_state for the initial transition.

• extra_columns (list[Column] | None) – Additional columns on the raw table (e.g. changed_by, notes).

• allow_initial (bool) – If True (default), a row with from_state = NULL is allowed as the first transition for a subject. Set to False to require an explicit initial state defined in valid_transitions.

Raises:

CodegenDatabaseValidationError – If states or valid_transitions is empty, or if a transition references an undeclared state.

can_transition_to(from_state, to_state)

Return whether from_state -> to_state is a legal transition.

Parameters:

• from_state (Any) – The subject’s current state, or None for a subject that hasn’t transitioned yet.

• to_state (Any) – The state being checked.

Return type:

bool

Returns:

True if (from_state, to_state) is in valid_transitions.

async transition_many(db, *, subjects, to_state, **extras)

Bulk-transition every subject in subjects to to_state.

Issues a single INSERT INTO raw (subject, from_state, to_state, ...) SELECT ... FROM current_view after locking the latest raw row per affected subject – mirroring what transition_to() does for a single subject so concurrent single-row and bulk callers serialize against each other.

The set of legal current states is auto-derived from the SM’s declared transitions: only subjects whose current state has a registered (from, to_state) edge are advanced. Subjects in any other state are silently skipped. This is the same set the BEFORE INSERT trigger would accept, applied as a WHERE filter so the trigger never has to reject rows.

The BEFORE INSERT trigger validates each emitted row, so a misconfigured to_state (one that’s not a valid target from any state) results in zero rows inserted – not an error.

AsyncSession only. Unlike transition_to() and call(), this method issues two statements (the FOR UPDATE lock and the INSERT ... SELECT) and awaits both internally – so it cannot dispatch transparently through a sync Session. If you need a sync variant, run the two statements manually with the helpers exposed on self.table and self.current_view.

Parameters:

• db (Any) – An AsyncSession.

• subjects (Select | Iterable[Any]) – A SQLAlchemy Select that yields subject ids, or any iterable of ids (UUIDs, ints, etc.). Iterables are bound as a VALUES list; a Select is correlated as a subquery so callers can compose against other tables (select(LineItem.id).where(...)).

• to_state (Any) – Target state. enum.Enum values are coerced via .value.

• **extras (Any) – Values for the SM’s extra columns (actor_id, note, …). Each value is applied to every row. Enum values are coerced.

Return type:

Any

Returns:

The Result from the INSERT ... RETURNING – iterate .scalars() for the subject ids that actually transitioned.

transition_to(db, **kwargs)

Invoke the SM’s generated transition_to SQL function.

Builds select(func.<schema>.<schema>_<name>_transition_to( ...)) and dispatches via db.execute(...). The SQL function locks the latest raw row for the subject, reads the actual current state, and inserts a new row with from_state = <current> – the BEFORE INSERT trigger then validates the (from_state, to_state) pair.

Kwargs are keyed by the subject column’s name (e.g. line_item_id for a state machine whose subject is a line item), to_state, plus any extra-column names declared on the SM. enum.Enum values are coerced to .value so callers can pass typed enums directly.

Parameters:

• db (Any) – Any object with an execute(stmt) method – Session, AsyncSession, or Connection. codegen_database does no awaiting itself; if execute returns an awaitable, await the return value at the call site.

• **kwargs (Any) – <subject_col_name>, to_state, and any extra-column kwargs declared on the SM.

Return type:

Any

Returns:

Whatever db.execute(stmt) returns – a Result for sync sessions, an Awaitable[Result] for async ones.

Raises:

KeyError – If the subject or to_state kwarg is missing.

class StateMachineCurrentStatePlugin(*, sm, column_type, column_name='status')

Maintain a denormalized, read-only current-state column.

class Deal(Base):

__plugins__ = [

UUIDV7PKPlugin(), StateMachineCurrentStatePlugin(

sm=deal_sm, column_type=DealStatus

),

]

status: Mapped[DealStatus]

Parameters:

• sm (CodegenDatabaseStateMachine) – The CodegenDatabaseStateMachine whose current state is mirrored onto the subject.

• column_type (type[Enum]) – The enum.Enum the SM’s states belong to.

• column_name (str) – Name of the current-state column on the subject (default "status").

construct_state_machine_current_query(sm)

Return a select against the current-state view.

Parameters:

sm (CodegenDatabaseStateMachine) – A CodegenDatabaseStateMachine instance.

Return type:

Select

Returns:

A SQLAlchemy Select over {name}_current.

construct_state_machine_history_query(sm, subject_id=None)

Return all transitions, optionally filtered to one subject.

Parameters:

• sm (CodegenDatabaseStateMachine) – A CodegenDatabaseStateMachine instance.

• subject_id (object) – When provided, adds a WHERE subject_col = subject_id filter.

Return type:

Select

Returns:

A SQLAlchemy Select.

Incremental refresh

Generic incremental refresh plugin.

IncrementalRefreshPlugin registers a checkpoint-based {schema}_{tablename}_refresh(p_since TIMESTAMPTZ DEFAULT NULL) function in PostgreSQL. Callers invoke this function from pg_cron (or manually after bulk loads) to roll forward an aggregate table.

The refresh function resolves the starting cutoff as:

_since = COALESCE(p_since, fallback_sql, '-infinity'::TIMESTAMPTZ)

where fallback_sql is a SQL expression (supplied by the caller) that reads the last-processed watermark, e.g.:

SELECT MIN(updated_at) - INTERVAL '1 second' FROM my_snapshot

The user-supplied body is PL/pgSQL that runs after _since is resolved. It typically aggregates rows from the source table where watermark_col >= _since and upserts results into an aggregate table.

Usage:

orders = CodegenDatabaseSimple(
    "orders",
    "app",
    metadata,
    schema_items=[
        Column("customer_id", Integer, nullable=False),
        Column("amount", Numeric, nullable=False),
    ],
    extra_plugins=[
        IncrementalRefreshPlugin(
            body="""
INSERT INTO app.orders_daily_agg (customer_id, day, total)
SELECT customer_id,
       date_trunc('day', created_at),
       SUM(amount)
FROM app.orders_raw
WHERE created_at >= _since
GROUP BY 1, 2
ON CONFLICT (customer_id, day)
DO UPDATE SET total = orders_daily_agg.total + EXCLUDED.total;
""",
            fallback_sql=(
                "SELECT MIN(day::timestamptz)"
                " - INTERVAL '1 second'"
                " FROM app.orders_daily_agg"
            ),
            schedule="*/5 * * * *",
        ),
    ],
)

# The above auto-registers a CronJob named "app.app_orders_refresh"
# that runs every 5 minutes.  Alembic autogenerate emits:
#   op.execute("SELECT cron.schedule(...)")
# Requires pg_cron to be installed.  Register it alongside the
# factory:
#   register_pg_extension(metadata, PGExtension("pg_cron", cascade=True))

class IncrementalRefreshPlugin(body, fallback_sql, *, schedule=None, table_key='raw_table')

Register a checkpoint-based incremental refresh function.

Generates and registers a PostgreSQL function with signature:

{schema}.{schema}_{tablename}_refresh(
    p_since TIMESTAMPTZ DEFAULT NULL
) RETURNS void

The function resolves _since as the caller-supplied p_since, or falls back to the expression in fallback_sql. The user-supplied body is embedded after _since is resolved and can reference _since directly.

Parameters:

• body (str) – PL/pgSQL fragment (no DECLARE or BEGIN/END wrapper). Must be valid PL/pgSQL and may reference _since TIMESTAMPTZ.

• fallback_sql (str) – SQL expression that returns a TIMESTAMPTZ used as the checkpoint when p_since is NULL. A NULL result falls back to '-infinity'. Typically reads MIN(updated_at) - INTERVAL '1 second' from the aggregate table.

• table_key (str) – Key in ctx for the source table (used only to satisfy the @requires dependency). Default "raw_table".

run(ctx)

Render and register the incremental refresh function.

Return type:

None

Query builders for incremental refresh patterns.

construct_unincorporated_since_query() returns a SELECT for rows in the source table that have not yet been rolled into an aggregate, based on a watermark timestamp.

Usage:

# Find orders not yet incorporated into the daily aggregate.
q = construct_unincorporated_since_query(
    orders,
    watermark_col="created_at",
    since="2024-06-01T00:00:00Z",
)

# Without a fixed timestamp: use a subquery as the checkpoint.
q = construct_unincorporated_since_query(
    orders,
    watermark_col="created_at",
    since_subquery="SELECT MIN(day::timestamptz) FROM app.orders_daily_agg",
)

construct_unincorporated_since_query(source, *, watermark_col='created_at', since=None, since_subquery=None, table_key='raw_table')

Return rows in source not yet incorporated into an aggregate.

Generates:

SELECT * FROM <source_table>
WHERE <watermark_col> >= <since>

where <since> is either a literal timestamp or the result of a subquery.

Parameters:

• source (ContextSource) – A factory instance or declarative class with a populated context.

• watermark_col (str) – Column in the source table to compare against the cutoff (default "created_at").

• since (str | None) – ISO 8601 timestamp string used as the lower bound. Exactly one of since or since_subquery must be given.

• since_subquery (str | None) – SQL expression (a scalar subquery) that returns a TIMESTAMPTZ cutoff. Exactly one of since or since_subquery must be given.

• table_key (str) – Key in ctx for the source table (default "raw_table").

Return type:

Select

Returns:

A SQLAlchemy Select.

Raises:

CodegenDatabaseValidationError – If neither or both of since and since_subquery are provided.

Cron

pg_cron extension for codegen_database.

PGCronExtension wires pg_cron support into the codegen_database lifecycle. Register it on a CodegenDatabaseConfig to enable declarative cron job scheduling via CronJob and IncrementalRefreshPlugin.

Usage:

from codegen_database.config import CodegenDatabaseConfig
from codegen_database.ext.pg_cron import PGCronExtension

config = CodegenDatabaseConfig()
config.use(PGCronExtension())

After registration, any CronJob stored in metadata (e.g. by IncrementalRefreshPlugin with a schedule argument) will be diffed against the live cron.job table during alembic revision --autogenerate. Alembic emits op.execute("SELECT cron.schedule(...)") for new or changed jobs.

The extension also auto-declares the pg_cron PostgreSQL extension in metadata so that the Alembic comparator can emit CREATE EXTENSION IF NOT EXISTS pg_cron CASCADE when needed.

Convenience re-exports:

from codegen_database.ext.pg_cron import (
    PGCronExtension,
    CronJob,
    register_cron_job,
    PGExtension,
    register_pg_extension,
)

class CronJob(name, schedule, command, database=None)

Describe a pg_cron job to register.

Parameters:

• name (str) – Unique job name used as the identity key in cron.job.

• schedule (str) – Cron expression, e.g. "*/5 * * * *".

• command (str) – SQL to execute, e.g. "SELECT app.app_orders_refresh()".

• database (str | None) – Database to run the command in. When None, the current database is used.

to_sql_schedule()

Render SELECT cron.schedule(...) DDL.

Return type:

str

Returns:

A complete SELECT cron.schedule(...) SQL string.

to_sql_unschedule()

Render SELECT cron.unschedule(...) DDL.

Return type:

str

Returns:

A complete SELECT cron.unschedule(...) SQL string.

class PGCronExtension(name='pg_cron')

Wire pg_cron support into the codegen_database lifecycle.

When registered on a CodegenDatabaseConfig:

• configure_metadata declares pg_cron as a required PostgreSQL extension so Alembic can create it when absent.

• configure_alembic registers the cron job comparator and renderer so alembic revision --autogenerate emits SELECT cron.schedule(...) for new or changed jobs, and the extension comparator so CREATE EXTENSION IF NOT EXISTS pg_cron CASCADE is emitted when pg_cron is missing.

Plugins that use pg_cron (e.g. IncrementalRefreshPlugin with a schedule argument) call assert_pg_extension_declared() to verify this extension is registered, giving a clear error message when it is absent.

Parameters:

name (str) – Extension name. Defaults to "pg_cron".

configure_alembic()

Register cron and extension Alembic comparators/renderers.

Return type:

None

configure_metadata(metadata)

Declare pg_cron as a required PostgreSQL extension.

Return type:

None

class PGExtension(name, schema=None, cascade=False)

Describe a PostgreSQL extension to install.

Parameters:

• name (str) – Extension name, e.g. "btree_gist".

• schema (str | None) – Schema in which to install the extension. When None the database default (usually public) is used.

• cascade (bool) – When True, adds CASCADE so dependent extensions are installed automatically.

to_sql_create()

Render CREATE EXTENSION IF NOT EXISTS DDL.

Return type:

str

Returns:

A complete CREATE EXTENSION SQL string.

register_cron_job(metadata, job)

Store job in metadata for Alembic autogenerate.

Parameters:

• metadata (MetaData) – SQLAlchemy MetaData to register on.

• job (CronJob) – The cron job to register.

Return type:

None

register_pg_extension(metadata, ext)

Store ext in metadata for Alembic autogenerate.

Parameters:

• metadata (MetaData) – SQLAlchemy MetaData to register on.

• ext (PGExtension) – The extension to register.

Return type:

None

pg_cron job dataclasses and metadata registration.

class CronJob(name, schedule, command, database=None)

Describe a pg_cron job to register.

Parameters:

• name (str) – Unique job name used as the identity key in cron.job.

• schedule (str) – Cron expression, e.g. "*/5 * * * *".

• command (str) – SQL to execute, e.g. "SELECT app.app_orders_refresh()".

• database (str | None) – Database to run the command in. When None, the current database is used.

to_sql_schedule()

Render SELECT cron.schedule(...) DDL.

Return type:

str

Returns:

A complete SELECT cron.schedule(...) SQL string.

to_sql_unschedule()

Render SELECT cron.unschedule(...) DDL.

Return type:

str

Returns:

A complete SELECT cron.unschedule(...) SQL string.

class CronJobs(jobs=<factory>)

Container for all cron jobs registered on a MetaData.

Stored under metadata.info["cron_jobs"] by register_cron_job().

classmethod extract(metadata)

Return registered cron jobs or None if none exist.

Parameters:

metadata (object) – SQLAlchemy MetaData, or None.

Return type:

CronJobs | None

Returns:

The CronJobs holder or None.

register_cron_job(metadata, job)

Store job in metadata for Alembic autogenerate.

Parameters:

• metadata (MetaData) – SQLAlchemy MetaData to register on.

• job (CronJob) – The cron job to register.

Return type:

None

Alembic comparator ops and diff logic for pg_cron jobs.

class ScheduleCronJobOp(job)

Alembic operation: create or update a pg_cron job.

reverse()

Return a no-op downgrade: jobs are never dropped automatically.

cron.schedule is an upsert so re-running it on downgrade is always safe. Users who want to remove a job on downgrade should add an explicit UnscheduleCronJobOp call.

Return type:

ScheduleCronJobOp

to_sql()

Return the SQL statement for this operation.

Return type:

list[str]

class UnscheduleCronJobOp(name)

Alembic operation: remove a pg_cron job.

This op is available for manual migration authoring. The autogenerate comparator never emits it automatically to avoid accidentally removing externally-managed jobs.

to_sql()

Return the SQL statement for this operation.

Return type:

list[str]

compare_cron_jobs(connection, desired)

Diff desired cron jobs against the current database state.

Only emits ScheduleCronJobOp (create or update) — jobs are never dropped automatically. Use UnscheduleCronJobOp in a hand-written migration to remove a job.

Parameters:

• connection (Connection) – Active database connection.

• desired (CronJobs) – The CronJobs holder read from metadata.

Return type:

list[ScheduleCronJobOp]

Returns:

List of ScheduleCronJobOp operations to bring declared jobs into sync with the database.

PostGIS

PostGIS extension for codegen_database.

PostGISExtension wires PostGIS-suite support into the codegen_database lifecycle. Register it on a CodegenDatabaseConfig to enable the PostGIS column types (currently STDADDR, the stdaddr composite produced by the address_standardizer extension).

Usage:

from codegen_database.config import CodegenDatabaseConfig
from codegen_database.ext.postgis import PostGISExtension

config = CodegenDatabaseConfig()
config.use(PostGISExtension())

PostGISExtension is the grouping for the whole PostGIS suite; which PostgreSQL extensions it actually installs are controlled by boolean flags. By default it declares only address_standardizer (which provides stdaddr), since that installs independently of the heavy postgis core. Enable the core when you need geometry:

config.use(PostGISExtension(postgis=True))

Enable the TIGER geocoder when you need to turn a free-form address string into ranked candidate addresses with coordinates (the engine behind fsh_lib.geocode.find_address_candidates()):

config.use(PostGISExtension(tiger_geocoder=True))

postgis_tiger_geocoder is declared CASCADE because it depends on postgis and fuzzystrmatch; CASCADE installs those automatically. (It does not pull in address_standardizer – that is a sibling extension for STDADDR, installed by the default address_standardizer flag above; the geocoder does not require it.) The extension only ships the geocode machinery – it does not load the TIGER census data the geocoder matches against. That data load (Loader_Generate_Nation_Script / Loader_Generate_Census_Script per state, US-only) is a separate operational step; geocode returns no rows until it has run. See tiger-loader/ for the loader image that performs it.

After registration the extension auto-declares the selected PostgreSQL extensions in metadata, so the Alembic comparator emits CREATE EXTENSION IF NOT EXISTS ... for any that are missing from the database. Columns then use the type directly:

from sqlalchemy import Column
from codegen_database.ext.postgis import STDADDR, StdAddr

Column("address", STDADDR(), nullable=True)
# store StdAddr(house_num="123", name="MAIN", suftype="ST")

Convenience re-exports:

from codegen_database.ext.postgis import (
    PostGISExtension,
    STDADDR,
    StdAddr,
    PGExtension,
    register_pg_extension,
)

class PostGISExtension(name='postgis', postgis=False, address_standardizer=True, geography=False, tiger_geocoder=False)

Wire PostGIS-suite support into the codegen_database lifecycle.

The extension groups the PostGIS suite; the boolean flags select which PostgreSQL extensions it installs.

Parameters:

• name (str) – Extension name. Defaults to "postgis".

• address_standardizer (bool) – Important for STDADDR

• geography (bool) – Important for COORDINATE

• postgis (bool) – Important for…… everything else

• tiger_geocoder (bool) – Important for free-string geocoding (geocode); CASCADE pulls in its dependencies.

configure_metadata(metadata)

Declare the selected PostgreSQL extensions.

Return type:

None

Row-level security

Alembic autogenerate integration for RLS policies.

Call register_rls_alembic_events() in env.py alongside alembic_hook() to enable RLS policy diffing in Alembic autogenerate:

from codegen_database.alembic.register import alembic_hook
from codegen_database.ext.rls.alembic import register_rls_alembic_events

alembic_hook()
register_rls_alembic_events()

The comparator queries pg_policies and emits op.execute(...) statements for any policies that need to be created or dropped.

register_rls_alembic_events()

Register RLS comparator and renderer into Alembic.

Call once in env.py before context.configure().

Return type:

None