Declarative style¶

codegen_database’s primary style for defining dimensions is declarative — subclass CodegenDatabaseBase once to create a project base, then define each model as a class with __tablename__ and optionally __factory__ / __plugins__. The plugin pipeline runs at class-definition time, the table is registered on the metadata, and the class is ORM-mapped so that select(MyModel) works directly.

An imperative style is also available for cases where it suits better — see Kitchen-sink imperative example below. Both styles produce identical SQLAlchemy tables, and Alembic migrations.

Quick comparison ¶

Same dimension, both styles:

# Declarative (preferred)
from codegen_database import CodegenDatabaseBase

class Base(CodegenDatabaseBase):
    metadata = metadata

class Products(Base):
    __tablename__ = "products"
    __table_args__ = {"schema": "public"}

    name = Column(String, nullable=False)
    price = Column(Integer, nullable=False)

# Imperative (alternative)
from codegen_database.factory import CodegenDatabaseSimple

products = CodegenDatabaseSimple(
    "products", "public", metadata,
    schema_items=[
        Column("name", String, nullable=False),
        Column("price", Integer, nullable=False),
    ],
)

After definition, both expose the same interface:

Products.ctx      # FactoryContext
Products.table    # SQLAlchemy Table
Products.id       # primary-key Column (InstrumentedAttribute after mapping)
Products.name     # InstrumentedAttribute — usable in select(), filter(), etc.

products.ctx      # FactoryContext
products.table    # SQLAlchemy Table

Downstream helpers accept both.

Using `mapped_column` and `Mapped[]`¶

codegen_database accepts both SQLAlchemy’s classic Column() form and the 2.0-style mapped_column() form. The Mapped[X] annotation is accepted purely for type checkers — it is not read at runtime.

from sqlalchemy import Integer, String
from sqlalchemy.orm import Mapped, mapped_column

class Users(Base):
    __tablename__ = "users"
    __table_args__ = {"schema": "public"}

    name: Mapped[str] = mapped_column(Text, nullable=False)
    age: Mapped[int | None] = mapped_column(Integer)

One rule: always pass an explicit SQL type to mapped_column(...).

SQLAlchemy’s own declarative metaclass can infer a column’s SQL type and nullability from a Mapped[X] annotation (Mapped[int] → Integer, Mapped[str | None] → nullable=True). codegen_database uses imperative mapping and does not run that inference pass, so the annotation is decorative at runtime. If you omit the type, codegen_database raises CodegenDatabaseValidationError rather than generating a broken column.

# OK — Mapped[] for type checkers, explicit type for codegen_database
name: Mapped[str] = mapped_column(Text)

# OK — classic form still works
name = Column(Text)

# ERROR — no SQL type to build the column from
name: Mapped[str] = mapped_column()

# ERROR — bare annotation with no assignment
name: Mapped[str]

Nullability, defaults, server defaults, foreign keys, etc. behave identically whether you use Column(...) or mapped_column(...); codegen_database uses the underlying Column object in both cases.

Declarative style in depth ¶

Setting up the base¶

Subclass CodegenDatabaseBase once to create your project base, then attach the shared MetaData and optional CodegenDatabaseConfig:

from sqlalchemy import MetaData
from codegen_database import CodegenDatabaseBase, CodegenDatabaseView, construct_naming_conventions_dict
from codegen_database.config import CodegenDatabaseConfig

metadata = MetaData(
    naming_convention=construct_naming_conventions_dict(),
)

config = CodegenDatabaseConfig(auto_discover=False)
metadata.info["codegen_database_config"] = config


class Base(CodegenDatabaseBase):
    metadata = metadata
    codegen_database_config = config


class ViewBase(CodegenDatabaseView):
    metadata = metadata

Dimension classes¶

Every subclass of Base that has a __tablename__ is treated as a model. codegen_database’s plugin pipeline runs at class-definition time — the class is immediately ORM-mapped and the table is registered on the metadata.

class Users(Base):
    __tablename__ = "users"
    __table_args__ = {"schema": "public"}

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

`factory` and `plugins`¶

Two optional per-model attributes configure the plugin pipeline:

__factory__ — the ResourceFactory subclass whose _INTERNAL_PLUGINS define the storage strategy (default CodegenDatabaseSimple).
__plugins__ — a list of plugin instances appended to the resolved plugin list.

from codegen_database.factory.dimension.append_only import CodegenDatabaseAppendOnly
from codegen_database.factory.dimension.eav import CodegenDatabaseEAV
from codegen_database.factory.ledger import CodegenDatabaseLedger

class Orders(Base):
    __tablename__ = "orders"
    __table_args__ = {"schema": "public"}
    # No __factory__ / __plugins__ needed — CodegenDatabaseSimple is the default.

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


class Students(Base):
    __tablename__ = "students"
    __table_args__ = {"schema": "private"}
    __factory__ = CodegenDatabaseAppendOnly

    name = Column(String)
    user_id = Column(Integer, CodegenDatabaseForeignKey("users.id"))


class Products(Base):
    __tablename__ = "products"
    __table_args__ = {"schema": "private"}
    __factory__ = CodegenDatabaseEAV

    color = Column(String)
    price = Column(Integer)


class Invoices(Base):
    __tablename__ = "invoices"
    __table_args__ = {"schema": "private"}
    __factory__ = CodegenDatabaseAppendOnly

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


class Inventory(Base):
    __tablename__ = "inventory"
    __table_args__ = {"schema": "private"}
    __factory__ = CodegenDatabaseLedger

    warehouse = Column(String, nullable=False)
    sku = Column(String, nullable=False)

Adding checks and constraints¶

CodegenDatabaseCheck and CodegenDatabaseIndex go in __table_args__ (matching standard SQLAlchemy declarative convention):

class Products(Base):
    __tablename__ = "products"
    __table_args__ = (
        CodegenDatabaseCheck("{price} > 0", name="positive_price"),
        CodegenDatabaseIndex("idx_products_color", "{color}"),
        {"schema": "public"},
    )

    color = Column(String)
    price = Column(Integer, nullable=False)

Declarative views¶

CodegenDatabaseView defines aggregate or join views. The class is ORM-mapped so that select(CustomerStats) works directly.

class CustomerStats(ViewBase):
    __tablename__ = "customer_order_stats"
    __table_args__ = {"schema": "public"}

    # Explicit columns give accurate types and are self-documenting.
    customer_id = Column(Integer)
    order_count = Column(Integer)
    order_total = Column(Numeric(10, 2))

    __query__ = select(
        Orders.customer_id,
        func.count().label("order_count"),
        func.sum(Orders.total).label("order_total"),
    ).group_by(Orders.customer_id)

Omitting explicit columns causes codegen_database to infer them from the query’s selected_columns (best-effort types, may fall back to NullType). Explicit columns are preferred.

To create a materialized view:

from codegen_database import ViewOptions

class ProductSummary(ViewBase):
    __tablename__ = "product_summary"
    __table_args__ = {"schema": "public"}
    __options__ = ViewOptions(materialized=True)

    __query__ = select(...)

Kitchen-sink declarative example ¶

A complete project showing all dimension types, views, constraints, foreign keys, and ledger events using the declarative style:

from sqlalchemy import (
    Column, Computed, Float, Integer, MetaData,
    Numeric, String, func, literal_column, select, union_all,
)
from sqlalchemy.dialects.postgresql import ARRAY

from codegen_database import (
    CodegenDatabaseBase, CodegenDatabaseForeignKey,
    CodegenDatabaseFunctionMixin, CodegenDatabaseViewMixin,
    construct_naming_conventions_dict,
)
from codegen_database.check import CodegenDatabaseCheck
from codegen_database.config import CodegenDatabaseConfig
from codegen_database.ext.ledger import (
    LedgerEvent,
    construct_ledger_balance_query,
    construct_ledger_balances_query,
    ledger_event_function,
)
from codegen_database.factory.dimension.append_only import CodegenDatabaseAppendOnly
from codegen_database.factory.dimension.eav import CodegenDatabaseEAV
from codegen_database.factory.ledger import CodegenDatabaseLedger
from codegen_database.plugins.ledger import DoubleEntryPlugin, DoubleEntryTriggerPlugin

metadata = MetaData(naming_convention=construct_naming_conventions_dict())
config = CodegenDatabaseConfig(auto_discover=False)
metadata.info["codegen_database_config"] = config

class Base(CodegenDatabaseBase):
    metadata = metadata
    codegen_database_config = config

# -- Dimension tables -------------------------------------------------------

class Users(Base):
    __tablename__ = "users"
    __table_args__ = (
        CodegenDatabaseCheck("{price} > 0", name="positive_price"),
        CodegenDatabaseCheck("{qty} >= 0", name="nonneg_qty"),
        {"schema": "public"},
    )
    name = Column(String)
    price = Column(Integer)
    qty = Column(Integer)
    total = Column(Integer, Computed("price * qty"))

class Students(Base):
    __tablename__ = "students"
    __table_args__ = {"schema": "private"}
    __factory__ = CodegenDatabaseAppendOnly
    name = Column(String)
    user_id = Column(Integer, CodegenDatabaseForeignKey("users.id"))

class Invoices(Base):
    __tablename__ = "invoices"
    __table_args__ = {"schema": "private"}
    __factory__ = CodegenDatabaseAppendOnly
    customer_id = Column(Integer, nullable=False)
    amount = Column(Numeric(10, 2), nullable=False)

class InvoiceLines(Base):
    __tablename__ = "invoice_lines"
    __table_args__ = {"schema": "private"}
    __factory__ = CodegenDatabaseEAV
    invoice_id = Column(Integer, CodegenDatabaseForeignKey("invoices.id"), nullable=False)
    department = Column(String, nullable=False)
    amount = Column(Integer, nullable=False)

class Orders(Base):
    __tablename__ = "orders"
    __table_args__ = {"schema": "public"}
    customer_id = Column(Integer, nullable=False)
    total = Column(Numeric(10, 2), nullable=False)
    internal_notes = Column(String, nullable=True)

class Customers(Base):
    __tablename__ = "customers"
    __table_args__ = {"schema": "public"}
    name = Column(String, nullable=False)
    email = Column(String)

# -- Views ------------------------------------------------------------------

class OrderStats(CodegenDatabaseViewMixin, Base):
    __tablename__ = "customer_order_stats"
    __table_args__ = {"schema": "public"}
    customer_id = Column(Integer)
    order_count = Column(Integer)
    order_total = Column(Numeric(10, 2))
    __query__ = select(
        Orders.customer_id,
        func.count().label("order_count"),
        func.sum(Orders.total).label("order_total"),
    ).group_by(Orders.customer_id)

# -- Inventory ledger -------------------------------------------------------

_adjust_event = LedgerEvent(
    name="adjust",
    input=lambda param: select(
        param("warehouse", String).label("warehouse"),
        param("sku", String).label("sku"),
        param("value", Integer).label("value"),
    ),
)

class Inventory(Base):
    __tablename__ = "inventory"
    __table_args__ = {"schema": "private"}
    __factory__ = CodegenDatabaseLedger
    warehouse = Column(String, nullable=False)
    sku = Column(String, nullable=False)

class InventoryBalances(CodegenDatabaseViewMixin, Base):
    __tablename__ = "inventory_balances"
    __table_args__ = {"schema": "private"}
    __query__ = construct_ledger_balance_query(Inventory, ["warehouse", "sku"])

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

# -- Double-entry ledger ----------------------------------------------------

_post_event = LedgerEvent(
    name="post",
    input=lambda param: select(
        func.unnest(param("invoice_ids", ARRAY(Integer))).label("invoice_id"),
    ),
    desired=lambda pginput: union_all(
        select(
            InvoiceLines.invoice_id, InvoiceLines.department,
            literal_column("'accounts_receivable'").label("account"),
            literal_column("'debit'").label("direction"),
            InvoiceLines.amount.label("value"),
        ).where(InvoiceLines.invoice_id.in_(select(pginput.c.invoice_id))),
        select(
            InvoiceLines.invoice_id, InvoiceLines.department,
            literal_column("'revenue'").label("account"),
            literal_column("'credit'").label("direction"),
            InvoiceLines.amount.label("value"),
        ).where(InvoiceLines.invoice_id.in_(select(pginput.c.invoice_id))),
    ),
    existing=construct_ledger_balances_query("invoice_id", "department", "account", "direction"),
    diff_keys=["invoice_id", "department", "account", "direction"],
)

class Ledger(Base):
    __tablename__ = "ledger"
    __table_args__ = {"schema": "private"}
    __factory__ = CodegenDatabaseLedger
    __plugins__ = [
        DoubleEntryPlugin(),
        DoubleEntryTriggerPlugin(),
    ]
    invoice_id = Column(Integer, nullable=False)
    department = Column(String, nullable=False)
    account = Column(String, nullable=False)

class LedgerBalances(CodegenDatabaseViewMixin, Base):
    __tablename__ = "construct_ledger_balances_query"
    __table_args__ = {"schema": "private"}
    __query__ = construct_ledger_balance_query(
        Ledger, ["invoice_id", "department", "account"]
    )

class PostInvoices(CodegenDatabaseFunctionMixin, Base):
    __table_args__ = {"schema": "private"}
    __funcspec__ = ledger_event_function(Ledger, _post_event)

Kitchen-sink imperative example ¶

The same schema defined with the imperative (factory-call) style. Columns are passed as schema_items; extra plugins go in extra_plugins. Use this style when you want minimal boilerplate or need to construct factories programmatically (e.g. looping over a config dict).

from sqlalchemy import Column, Integer, MetaData, Numeric, String
from codegen_database.factory import CodegenDatabaseSimple
from codegen_database.factory.dimension.append_only import CodegenDatabaseAppendOnly
from codegen_database.factory.dimension.eav import CodegenDatabaseEAV
from codegen_database.factory.ledger import CodegenDatabaseLedger
from codegen_database.check import CodegenDatabaseCheck
from codegen_database import construct_naming_conventions_dict
from codegen_database.fk import CodegenDatabaseForeignKey
from codegen_database.functions import CodegenDatabaseFunction
from codegen_database.ext.ledger import construct_ledger_balance_query, ledger_event_function
from codegen_database.views import CodegenDatabasePlainView

metadata = MetaData(naming_convention=construct_naming_conventions_dict())

users = CodegenDatabaseSimple(
    "users", "public", metadata,
    schema_items=[
        Column("name", String),
        Column("price", Integer),
        Column("qty", Integer),
        CodegenDatabaseCheck("{price} > 0", name="positive_price"),
    ],
)

students = CodegenDatabaseAppendOnly(
    "students", "private", metadata,
    schema_items=[
        Column("name", String),
        Column("user_id", Integer, CodegenDatabaseForeignKey("users.id")),
    ],
)

invoices = CodegenDatabaseAppendOnly(
    "invoices", "private", metadata,
    schema_items=[
        Column("customer_id", Integer, nullable=False),
        Column("amount", Numeric(10, 2), nullable=False),
    ],
)

orders = CodegenDatabaseSimple(
    "orders", "public", metadata,
    schema_items=[
        Column("customer_id", Integer, nullable=False),
        Column("total", Numeric(10, 2), nullable=False),
    ],
)

customers = CodegenDatabaseSimple(
    "customers", "public", metadata,
    schema_items=[
        Column("name", String, nullable=False),
        Column("email", String),
    ],
)

inventory = CodegenDatabaseLedger(
    "inventory", "private", metadata,
    schema_items=[
        Column("warehouse", String, nullable=False),
        Column("sku", String, nullable=False),
    ],
)

CodegenDatabasePlainView(
    name="inventory_balances",
    schema="private",
    metadata=metadata,
    query=construct_ledger_balance_query(inventory, ["warehouse", "sku"]),
)

_spec = ledger_event_function(inventory, _adjust_event)
CodegenDatabaseFunction(
    _spec["name"],
    inventory.ctx.schemaname,
    _spec["definition"],
    metadata=metadata,
    language=_spec["language"],
    parameters=_spec["parameters"],
    returns=_spec["returns"],
    security=_spec["security"],
)

Mixing styles ¶

Both styles share the same ContextSource duck-type interface (source.ctx), so you can freely mix them. Imperative factories and declarative classes are interchangeable as source arguments to construct_ledger_balance_query() and ledger_event_function().

# Imperative factory
orders = CodegenDatabaseSimple("orders", "public", metadata, schema_items=[...])

# Declarative class — accesses the same ctx
class Customers(Base):
    __tablename__ = "customers"
    __table_args__ = {"schema": "public"}
    name = Column(String)

Choosing a style ¶

Declarative is the default choice. It gives you:

Python classes that are directly queryable via select(MyModel)
ORM-mapped attributes (MyModel.name, MyModel.id) usable in where() and select() expressions without .table.c. indirection
Co-location of column definitions with the class that uses them
Familiarity for developers coming from SQLAlchemy declarative

Reach for the imperative style only when it’s a clearly better fit:

Programmatic construction (e.g. looping over a config dict to emit many near-identical tables)
Migration-only workflows where no ORM mapping is desired

Both styles generate identical Alembic migrations.

codegen_database

Navigation

Related Topics

Declarative style¶

Quick comparison ¶

Using `mapped_column` and `Mapped[]`¶

Declarative style in depth ¶

Setting up the base¶

Dimension classes¶

`factory` and `plugins`¶

Adding checks and constraints¶

Declarative views¶

Kitchen-sink declarative example ¶

Kitchen-sink imperative example ¶

Mixing styles ¶

Choosing a style ¶

Declarative style¶

Setting up the base¶

Dimension classes¶

__factory__ and __plugins__¶

Adding checks and constraints¶

Declarative views¶

`factory` and `plugins`¶