"""Declarative base classes for codegen_database-managed models and views.
Subclass :class:`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 :class:`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.
"""
from __future__ import annotations
import annotationlib
import enum
import inspect
import typing
from dataclasses import dataclass
from typing import TYPE_CHECKING, Any, ClassVar, Literal, overload
from sqlalchemy import Column, Table, func, select
from sqlalchemy import MetaData as _SAMetaData
from sqlalchemy import inspect as sa_inspect
from sqlalchemy import types as sa_types
from sqlalchemy.orm import Mapped, MappedColumn, MapperProperty
from sqlalchemy.orm import registry as SARegistry # noqa: N812
from sqlalchemy_declarative_extensions import (
View,
register_function,
register_view,
)
from sqlalchemy_declarative_extensions.dialects.postgresql import (
Function,
FunctionSecurity,
FunctionVolatility,
)
from codegen_database.errors import CodegenDatabaseValidationError
from codegen_database.factory.base import (
_resolve_plugins,
_run_plugin_validators,
_sort_plugins,
)
from codegen_database.factory.context import FactoryContext
from codegen_database.factory.dimension.simple import CodegenDatabaseSimple
from codegen_database.fk import DimensionRef, register_dimension
from codegen_database.utils.naming_convention import (
construct_naming_conventions_dict,
)
from codegen_database.utils.query import compile_query
if TYPE_CHECKING:
from sqlalchemy import Select
[docs]
@dataclass(frozen=True)
class ViewOptions:
"""Options for declarative views.
Set on a :class:`CodegenDatabaseViewMixin` or
:class:`CodegenDatabaseView` subclass via ``__options__``.
Attributes:
materialized: If ``True``, creates a PostgreSQL materialized
view instead of a regular view.
"""
materialized: bool = False
@overload
def _parse_table_args(
cls: type, *, schema_required: Literal[True]
) -> tuple[str, list]: ...
@overload
def _parse_table_args(
cls: type, *, schema_required: Literal[False] = False
) -> tuple[str | None, list]: ...
def _parse_table_args(
cls: type,
*,
schema_required: bool = False,
) -> tuple[str | None, list]:
"""Extract schema and extra constraints from __table_args__.
Args:
cls: The class whose ``__table_args__`` to read.
schema_required: Raise if no schema is declared, rather than
returning ``None``.
Returns:
``(schema, extra_constraints)`` where *schema* may be
``None`` unless *schema_required*, and *extra_constraints* is a
list of table-level objects (``CodegenDatabaseCheck``,
``ForeignKeyConstraint``, etc.).
Raises:
CodegenDatabaseValidationError: If *schema_required* and *cls*
declares no schema.
"""
schema, extra_constraints = _read_table_args(cls)
if schema_required and schema is None:
msg = (
f"{cls.__name__} must specify a schema via "
f"__table_args__ = {{'schema': '...'}} "
f"or as the final dict in a tuple __table_args__."
)
raise CodegenDatabaseValidationError(msg)
return schema, extra_constraints
def _read_table_args(cls: type) -> tuple[str | None, list]:
raw = getattr(cls, "__table_args__", None)
if raw is None:
return None, []
if isinstance(raw, dict):
return raw.get("schema"), []
if isinstance(raw, tuple) and raw:
items = list(raw)
last = items[-1]
if isinstance(last, dict):
return last.get("schema"), items[:-1]
return None, items
return None, []
def _mapped_column_to_column(attr_name: str, mc: MappedColumn) -> Column:
"""Extract a finalised :class:`~sqlalchemy.Column` from a ``MappedColumn``.
``mapped_column(...)`` is SQLAlchemy 2.0's typed replacement for
``Column()``. Its underlying ``Column`` is created eagerly from the
positional/keyword args, but the *name* is only assigned by SA's
declarative metaclass. codegen_database does not use that metaclass, so we
assign the name from the attribute here and reject declarations that
rely on ``Mapped[X]`` annotation-driven type inference (codegen_database has
no access to that inference pass).
Raises:
CodegenDatabaseValidationError: If ``mc`` has no explicit SQL type
(i.e. ``mapped_column()`` was called with no type argument).
"""
col = mc.column
if isinstance(col.type, sa_types.NullType):
msg = (
f"{attr_name!r}: mapped_column() requires an explicit "
f"SQL type in codegen_database. Annotation-driven inference "
f"(Mapped[int] -> Integer) is not supported. "
f"Use e.g. mapped_column(Integer) or mapped_column(String(100))."
)
raise CodegenDatabaseValidationError(msg)
# MappedColumn defers name/key assignment until the declarative
# metaclass runs; getattr sidesteps stub types that assume they are
# already strings.
if getattr(col, "name", None) is None:
col.name = attr_name
if getattr(col, "key", None) is None:
col.key = attr_name
return col
def _collect_columns(cls: type) -> list[Column]:
"""Read ``Column`` and ``mapped_column`` declarations from *cls*.
Accepts both:
- plain ``Column(...)`` assignments, and
- ``mapped_column(...)`` assignments (optionally annotated with
``Mapped[X]`` for type checkers).
Column objects whose ``name`` is ``None`` get the attribute
name assigned automatically (matching declarative behaviour).
Mixin columns are picked up too: every class in *cls*'s MRO that
isn't ``object``, ``cls`` itself, or a codegen_database ORM base contributes
its own column declarations. This is what lets users share
column sets via plain Python mixins (e.g. ``FileMixin``). Walking
the MRO bottom-up and skipping anything whose own dict already
holds a SQLAlchemy ``registry`` keeps inheritance chains rooted at
a codegen_database base from double-emitting columns the base already owns.
"""
sources: list[type] = [cls]
# Mixins in MRO order. Skip object, the class itself, and any
# ancestor that registers an ORM registry -- those are
# codegen_database-managed bases whose columns get owned at the base level
# and shouldn't be re-emitted on every concrete model.
sources.extend(
ancestor
for ancestor in cls.__mro__[1:]
if ancestor is not object and "_registry" not in ancestor.__dict__
)
columns: list[Column] = []
seen: set[str] = set()
for src in sources:
for attr_name, value in list(src.__dict__.items()):
if attr_name in seen:
continue
if isinstance(value, Column):
if value.name is None:
value.name = attr_name
if value.key is None:
value.key = attr_name
columns.append(value)
seen.add(attr_name)
elif isinstance(value, MappedColumn):
columns.append(_mapped_column_to_column(attr_name, value))
seen.add(attr_name)
return columns
def _check_no_bare_mapped_annotations(
cls: type,
*,
plugin_columns: frozenset[str],
) -> None:
"""Reject ``Mapped[X]`` annotations that have no assigned value.
In SQLAlchemy's own declarative base, writing ``name: Mapped[str]``
with no right-hand side is shorthand for ``mapped_column()`` and the
metaclass infers the column from the annotation. codegen_database uses
imperative mapping and does not run that inference, so a bare
annotation would silently produce no column. Surface this as a
clear error instead.
Annotations whose name appears in *plugin_columns* are accepted
without an assignment: those columns are produced by a configured
plugin (e.g. :class:`UUIDV4PKPlugin` injecting ``id``) and the
bare ``Mapped[X]`` annotation is the type-checker shim that
surfaces the right type without re-declaring the column itself.
SQLAlchemy's imperative mapper sets an ``InstrumentedAttribute``
on the class for every column on the resolved table, so the
annotation lines up with the runtime attribute SA installs --
no synthetic column is created from the annotation.
Only the class's own annotations dict is inspected.
"""
# Read annotations as *strings* (PEP 649 ``Format.STRING``) rather than
# evaluating them. The default ``Format.VALUE`` evaluates each annotation
# in the model's module namespace, which raises ``NameError`` for
# type-checking-only names -- e.g. a model that imports ``Mapped`` under
# ``if TYPE_CHECKING:`` and writes ``id: Mapped[uuid.UUID]``. We only need
# to detect the textual ``Mapped[...]`` form, so strings suffice and avoid
# evaluation entirely.
annotations: dict[str, Any] = inspect.get_annotations(
cls, format=annotationlib.Format.STRING
)
class_dict = cls.__dict__
bare: list[str] = []
for attr_name, annotation in annotations.items():
if isinstance(annotation, str):
is_mapped = "Mapped[" in annotation
else:
is_mapped = typing.get_origin(annotation) is Mapped
if (
is_mapped
and attr_name not in class_dict
and attr_name not in plugin_columns
):
bare.append(attr_name)
if bare:
attrs = ", ".join(repr(a) for a in bare)
msg = (
f"{cls.__name__}: Mapped[...] annotation without an "
f"assigned value on {attrs}. codegen_database does not support "
f"annotation-driven column inference; assign "
f"mapped_column(...) (with an explicit SQL type) or "
f"Column(...) explicitly."
)
raise CodegenDatabaseValidationError(msg)
def _collect_mapper_properties(cls: type) -> dict[str, MapperProperty]:
"""Read ORM ``MapperProperty`` declarations from ``cls.__dict__``.
Imperative mapping does not scan the class body the way
declarative mapping does, so attributes like
``tasks = relationship(Task)`` would be silently dropped.
Forwarding them to ``map_imperatively(properties=...)`` makes
class-body ``relationship()``, ``column_property()``, and
``synonym()`` declarations behave as users expect.
Only the class's own dict is searched — inherited properties
are left to SQLAlchemy's mapper inheritance.
"""
return {
attr_name: value
for attr_name, value in list(cls.__dict__.items())
if isinstance(value, MapperProperty)
}
def _find_registry(cls: type) -> SARegistry:
"""Return the ``_registry`` from the nearest base class in the MRO."""
for klass in cls.__mro__:
if "_registry" in klass.__dict__:
return klass.__dict__["_registry"]
msg = (
f"{cls.__name__}: no ORM registry found on any base class. "
"Subclass CodegenDatabaseBase (or CodegenDatabaseView) before "
"declaring models."
)
raise CodegenDatabaseValidationError(msg)
[docs]
def map_model(cls: type, parent_cls: type | None = None) -> None:
"""Run the codegen_database plugin pipeline for *cls* and ORM-map it.
Called from :meth:`CodegenDatabaseBase.__init_subclass__` once per
model class (those that have a ``__tablename__``), and ad hoc
for classes declared outside that hierarchy.
Args:
cls: The model class being defined.
parent_cls: 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.
"""
parent_cls = parent_cls if parent_cls is not None else cls
registry = _find_registry(parent_cls)
metadata = parent_cls.metadata
_validate_declaration(cls)
schema, extra_constraints = _parse_table_args(cls, schema_required=True)
factory_cls = getattr(cls, "__factory__", CodegenDatabaseSimple)
extra_plugins = list(getattr(cls, "__plugins__", []))
config = getattr(
parent_cls, "codegen_database_config", None
) or metadata.info.get("codegen_database_config")
user_columns = _collect_columns(cls)
schema_items: list = [
*user_columns,
*extra_constraints,
]
internal = list(factory_cls._INTERNAL_PLUGINS)
resolved = _resolve_plugins(config, None, extra_plugins, [], internal)
_run_plugin_validators(resolved)
ctx = FactoryContext(
tablename=cls.__tablename__,
schemaname=schema,
metadata=metadata,
schema_items=schema_items,
plugins=resolved,
)
for p in _sort_plugins(resolved):
p.run(ctx)
if "__root__" not in ctx:
msg = (
f"{cls.__name__}: no plugin produced '__root__'. "
f"Ensure the factory has a table-creating plugin."
)
raise CodegenDatabaseValidationError(msg)
# Validate bare ``Mapped[X]`` annotations now that we know which
# columns the plugin pipeline produced. Annotations whose name
# lands on the resolved table are accepted as type-checker shims
# for plugin-injected columns (e.g. ``id`` from
# :class:`UUIDV4PKPlugin`); any other bare ``Mapped[...]`` is
# still a misconfig because codegen_database has no annotation-driven
# inference pass to materialise the column from it.
_check_no_bare_mapped_annotations(
cls,
plugin_columns=frozenset(ctx["__root__"].c.keys()),
)
cls.__table__ = ctx["__root__"]
mapper_properties = _collect_mapper_properties(cls)
if mapper_properties:
registry.map_imperatively(
cls, cls.__table__, properties=mapper_properties
)
else:
registry.map_imperatively(cls, cls.__table__)
# Expose ctx and table so that view factories and query builders
# (construct_ledger_balance_query, ledger_event_function) that accept either
# a ResourceFactory instance or a declarative class can access
# them uniformly.
cls.ctx = ctx
cls.table = ctx["__root__"]
# Register the dimension so that CodegenDatabaseForeignKey can resolve
# references to this model by name (mirrors ResourceFactory.__init__).
fk_target_key: str = factory_cls._FK_TARGET_KEY
if fk_target_key in ctx:
fk_table = ctx[fk_target_key]
register_dimension(
metadata,
cls.__tablename__,
DimensionRef(schema=schema, table=fk_table.name),
)
[docs]
class CodegenDatabaseViewMixin:
"""Mixin that marks a :class:`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"))
"""
def _coerce_arg(value: Any) -> Any: # noqa: ANN401
"""Coerce a Python value for binding to a SQL function arg.
Currently unwraps :class:`enum.Enum` instances to their
``.value`` -- enough to let callers pass ``LineItemStatus.ORDERED``
where a ``text`` param is expected. Other types pass through
unchanged.
"""
return value.value if isinstance(value, enum.Enum) else value
[docs]
class CodegenDatabaseFunctionMixin:
"""Mark a :class:`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
:class:`~codegen_database.functions.CodegenDatabaseFunctionSpec`
(returned by
:func:`~codegen_database.ext.ledger.functions.ledger_event_function`,
chart-function builders, etc.). The schema is taken from
``__table_args__``, not from the spec.
Subclasses gain a :meth:`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,
)
"""
[docs]
@classmethod
def call(cls, db: Any, **kwargs: Any) -> Any: # noqa: ANN401
"""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); :class:`enum.Enum` values are
coerced to ``.value`` so callers can pass typed enums
directly.
Args:
db: 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: Function arguments keyed by param name
(without ``p_`` prefix).
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.
"""
spec: dict[str, Any] = cls.__dict__["__funcspec__"]
schema, _ = _parse_table_args(cls, schema_required=True)
fn_ref = getattr(getattr(func, schema), spec["name"])
args = [
_coerce_arg(kwargs[p.name.removeprefix("p_")])
for p in spec.get("parameters", [])
]
return db.execute(select(fn_ref(*args)))
def _build_function_model(cls: type) -> None:
"""Register a declarative function on the class's metadata.
Called from :meth:`CodegenDatabaseBase.__init_subclass__` for classes
that include :class:`CodegenDatabaseFunctionMixin`.
Args:
cls: The function class being defined. Must declare
``__funcspec__`` (a
:class:`~codegen_database.functions.CodegenDatabaseFunctionSpec`).
Raises:
AttributeError: If no base class in the MRO defines ``metadata``.
CodegenDatabaseValidationError: On missing schema.
"""
metadata = cls.metadata
_validate_declaration(cls, require_tablename=False)
schema, _ = _parse_table_args(cls, schema_required=True)
spec: dict[str, Any] = cls.__dict__["__funcspec__"]
fn = Function(
spec["name"],
spec["definition"],
returns=spec.get("returns", "void"),
language=spec.get("language", "sql"),
schema=schema,
parameters=spec.get("parameters", []),
security=spec.get("security", FunctionSecurity.invoker),
volatility=spec.get("volatility", FunctionVolatility.VOLATILE),
)
register_function(metadata, fn)
cls.function = fn
[docs]
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
:class:`~codegen_database.factory.dimension.simple.CodegenDatabaseSimple`).
Set to e.g. ``CodegenDatabaseAppendOnly`` for append-only semantics.
- ``__plugins__``: List of plugin instances appended to the
resolved plugin list.
"""
codegen_database_config: ClassVar[object | None] = None
# The PK column, injected at map time by the configured PK plugin (e.g.
# UUIDV4PKPlugin). Typed as Mapped[Any] -- the instance-level shim SA's
# imperative mapper backs with an InstrumentedAttribute -- so a model can
# narrow it (id: Mapped[uuid.UUID]) as a Mapped->Mapped override. (A
# ClassVar[Column] here instead clashed with that narrowing.) Inspected
# only on each subclass's own annotations, never the base's, so this bare
# annotation creates no column.
id: Mapped[Any]
# Set by map_model after the plugin pipeline runs.
__table__: ClassVar[Table]
table: ClassVar[Table]
ctx: ClassVar[FactoryContext]
# Set by __init_subclass__ on each direct subclass.
_registry: ClassVar[SARegistry]
metadata: ClassVar[_SAMetaData]
def __init_subclass__(cls, **kwargs: Any) -> None: # noqa: ANN401
"""Run the plugin pipeline when a model class is defined."""
super().__init_subclass__(**kwargs)
# Direct subclasses of CodegenDatabaseBase are base classes (e.g. the
# user's ``Base``). Give each its own ORM registry and
# auto-create metadata if none was provided.
if CodegenDatabaseBase in cls.__bases__:
cls._registry = SARegistry()
if "metadata" not in cls.__dict__:
cls.metadata = _SAMetaData(
naming_convention=construct_naming_conventions_dict()
)
return
# Functions are identified by CodegenDatabaseFunctionMixin — they don't
# need __tablename__ so handle them before that check.
if CodegenDatabaseFunctionMixin in cls.__mro__:
_build_function_model(cls)
return
if CodegenDatabaseViewMixin in cls.__mro__:
_build_view_model(cls, _find_registry(cls))
else:
map_model(cls)
def _validate_declaration(
cls: type,
*,
require_tablename: bool = True,
) -> None:
"""Reject *cls* if it is not a declarable codegen_database class.
Args:
cls: The class being declared.
require_tablename: Whether ``__tablename__`` is mandatory.
``False`` for functions, which are named by their
``__funcspec__`` instead.
Raises:
CodegenDatabaseValidationError: If *cls* is already mapped, or
declares no ``__tablename__``.
"""
if sa_inspect(cls, raiseerr=False) is not None:
msg = f"{cls.__name__}: already mapped."
raise CodegenDatabaseValidationError(msg)
if require_tablename and not hasattr(cls, "__tablename__"):
msg = (
f"{cls.__name__}: missing __tablename__. Declare one, or "
f"put shared columns on a plain mixin rather than a base "
f"subclass."
)
raise CodegenDatabaseValidationError(msg)
def _build_view_model(cls: type, registry: SARegistry) -> None:
"""Register a declarative view and ORM-map *cls* to it.
Called from :meth:`CodegenDatabaseBase.__init_subclass__` (and
:meth:`CodegenDatabaseView.__init_subclass__` for backward compatibility)
for classes that have both ``__tablename__`` and ``__query__``.
Columns are taken from the class's own ``Column`` declarations if
any are present (preferred — gives accurate types and makes intent
explicit). When no columns are declared the column list is derived
automatically from ``__query__.selected_columns``. In that case a
surrogate primary key is chosen for the ORM mapper: the column
named ``id`` if it exists, otherwise the first column.
Args:
cls: The view class being defined.
registry: The SQLAlchemy ``registry`` to use for imperative
mapping.
Raises:
AttributeError: If no base class in the MRO defines ``metadata``.
CodegenDatabaseValidationError: On missing schema or missing
``__query__``.
"""
metadata = cls.metadata
_validate_declaration(cls)
schema, _ = _parse_table_args(cls, schema_required=True)
query: Select | None = getattr(cls, "__query__", None)
if query is None:
msg = f"{cls.__name__}: must define __query__ = select(...)."
raise CodegenDatabaseValidationError(msg)
# Views don't run the plugin pipeline, so there's no auto-injected
# column set to widen the rule against -- a bare ``Mapped[X]`` on
# a view class is always a config error.
_check_no_bare_mapped_annotations(cls, plugin_columns=frozenset())
options: ViewOptions = cls.__dict__.get("__options__", ViewOptions())
# Columns: explicit declarations on the class take priority.
# Fallback: infer names and types from the query's selected columns.
explicit = _collect_columns(cls)
if explicit:
proxy_cols = explicit
else:
proxy_cols = [
Column(c.key, getattr(c, "type", sa_types.NullType()))
for c in query.selected_columns
if c.key is not None
]
# Proxy table on a private MetaData — Alembic sees only the view
# DDL (registered below), not this table.
tablename: str = cls.__tablename__
proxy = Table(tablename, _SAMetaData(), *proxy_cols, schema=schema)
# Register the view DDL on the main metadata for Alembic.
register_view(
metadata,
View(
tablename,
compile_query(query),
schema=schema,
materialized=options.materialized,
),
)
cls.__table__ = proxy
# SQLAlchemy requires at least one primary key column for ORM
# mapping. Views have no natural PK, so designate a surrogate
# unless the user already marked one with primary_key=True.
# Prefer a column named "id"; fall back to the first column.
mapper_kwargs: dict[str, Any] = {}
if not any(c.primary_key for c in proxy.c):
pk_name = next(
(c.name for c in proxy.c if c.name == "id"),
next(iter(proxy.c), Column("")).name,
)
pk_col = proxy.c.get(pk_name)
if pk_col is not None:
mapper_kwargs["primary_key"] = [pk_col]
mapper_properties = _collect_mapper_properties(cls)
if mapper_properties:
mapper_kwargs["properties"] = mapper_properties
registry.map_imperatively(cls, proxy, **mapper_kwargs)
[docs]
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.
"""
# Set by _build_view_model after the view is registered.
__table__: ClassVar[Table]
ctx: ClassVar[FactoryContext]
# Set by __init_subclass__ on each direct subclass.
_registry: ClassVar[SARegistry]
metadata: ClassVar[_SAMetaData]
def __init_subclass__(cls, **kwargs: Any) -> None: # noqa: ANN401
"""Register and ORM-map the view when the class is defined.
Raises:
CodegenDatabaseValidationError: If the subclass declares no
``__tablename__``.
"""
super().__init_subclass__(**kwargs)
# Direct subclasses of CodegenDatabaseView are project-level bases
# (e.g. the user's ``ViewBase``). Give each its own registry.
# metadata must be set explicitly — views belong to the same
# schema as their source tables and must share metadata with
# the model base so that Alembic sees everything in one place.
if CodegenDatabaseView in cls.__bases__:
cls._registry = SARegistry()
return
registry = _find_registry(cls)
_build_view_model(cls, registry)