Declarative style ================== codegen_database's primary style for defining dimensions is **declarative** — subclass :class:`~codegen_database.declarative.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 :ref:`imperative-style` below. Both styles produce identical SQLAlchemy tables, and Alembic migrations. .. contents:: :local: :depth: 1 Quick comparison ---------------- Same dimension, both styles: .. code-block:: python # 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) .. code-block:: python # 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: .. code-block:: python 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. .. _mapped-column-support: 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. .. code-block:: python 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 :class:`~codegen_database.errors.CodegenDatabaseValidationError` rather than generating a broken column. .. code-block:: python # 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 :class:`~codegen_database.declarative.CodegenDatabaseBase` once to create your project base, then attach the shared ``MetaData`` and optional :class:`~codegen_database.config.CodegenDatabaseConfig`: .. code-block:: python 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. .. code-block:: python 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 :class:`~codegen_database.factory.base.ResourceFactory` subclass whose ``_INTERNAL_PLUGINS`` define the storage strategy (default :class:`~codegen_database.factory.dimension.simple.CodegenDatabaseSimple`). - ``__plugins__`` — a list of plugin instances appended to the resolved plugin list. .. code-block:: python 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 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ :class:`~codegen_database.check.CodegenDatabaseCheck` and :class:`~codegen_database.index.CodegenDatabaseIndex` go in ``__table_args__`` (matching standard SQLAlchemy declarative convention): .. code-block:: python 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 ~~~~~~~~~~~~~~~~~ :class:`~codegen_database.declarative.CodegenDatabaseView` defines aggregate or join views. The class is ORM-mapped so that ``select(CustomerStats)`` works directly. .. code-block:: python 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**: .. code-block:: python 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: .. code-block:: python 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) .. _imperative-style: 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). .. code-block:: python 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 :func:`~codegen_database.ext.ledger.queries.construct_ledger_balance_query` and :func:`~codegen_database.ext.ledger.functions.ledger_event_function`. .. code-block:: python # 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.