"""Simple dimension resource factory."""
from __future__ import annotations
from pathlib import Path
from typing import TYPE_CHECKING, ClassVar
from sqlalchemy import Column, DateTime, Table, column, func, select
from sqlalchemy.schema import DefaultClause, FetchedValue
from codegen_database.factory.base import ResourceFactory
from codegen_database.factory.dimension._server_default import (
coalesced_value,
literal_defaults,
)
from codegen_database.plugin import (
Plugin,
PluginOrCollection,
produces,
requires,
singleton,
)
from codegen_database.plugins.check import TableCheckPlugin
from codegen_database.plugins.fk import TableFKPlugin
from codegen_database.plugins.index import TableIndexPlugin
from codegen_database.plugins.protect import RawTableProtectionPlugin
from codegen_database.plugins.soft_delete import soft_delete_columns
from codegen_database.plugins.trigger import (
InsteadOfTriggerPlugin,
TriggerOp,
render_delete_op,
)
from codegen_database.plugins.view import ViewPlugin
from codegen_database.utils.naming import resolve_name
from codegen_database.utils.query import compile_query
from codegen_database.utils.template import load_template
if TYPE_CHECKING:
from collections.abc import Callable
from sqlalchemy.sql.elements import ColumnClause
from codegen_database.factory.context import FactoryContext
_TEMPLATES = (
Path(__file__).resolve().parents[2] / "plugins" / "templates" / "simple"
)
_NAMING_DEFAULTS = {
"simple_raw_table": "%(table_name)s_raw",
"simple_function": "%(schema)s_%(table_name)s_%(op)s",
"simple_trigger": "%(schema)s_%(table_name)s_%(op)s",
}
def _build_returning(
primary: Table,
ctx: FactoryContext,
columns: list[str] | None,
dim_cols: list[str],
) -> str:
"""Render the ``RETURNING`` column list for the trigger.
Always an explicit list (never ``*``). ``RETURNING *`` expands to
the raw table's columns in *physical* order, but ``INTO NEW`` assigns
positionally into the *view* rowtype -- so if a later migration
``ALTER``-adds a column to the raw table (appended physically) while
the view places it in model order, ``*`` mis-binds and Postgres
raises a type mismatch. Naming the columns explicitly makes the
assignment order-independent of the raw table's physical layout.
The names come from ``primary`` (the raw table) in model order,
which for a simple passthrough view is exactly the view's column
order. The subset path lists the same set explicitly for the same
reason.
"""
if columns is None:
return ", ".join(c.name for c in primary.columns)
pk_name = ctx.pk_column_name
ret_cols = [pk_name, *dim_cols]
ret_cols.extend(
ctx[key]
for key in ("created_at_column", "updated_at_column")
if key in ctx and ctx[key] not in dim_cols
)
return ", ".join(ret_cols)
def _build_simple_ops_with_columns(
columns: list[str] | None,
table_key: str = "raw_table",
) -> Callable[[FactoryContext], list[TriggerOp]]:
"""Return an ops_builder that respects a column subset.
Args:
columns: Writable columns for the triggers.
When ``None``, uses all dim columns from ctx.
table_key: Key in ``ctx`` for the backing raw table.
Returns:
A callable suitable for ``InsteadOfTriggerPlugin``.
"""
def builder(ctx: FactoryContext) -> list[TriggerOp]:
primary = ctx[table_key]
base_fullname = f"{ctx.schemaname}.{primary.name}"
dim_cols = ctx.dim_column_names
auto_managed = ctx.get("exclude_from_mutations", set())
insert_cols = [col for col in dim_cols if col not in auto_managed]
# Columns with a literal server_default: when the view insert
# omits one, NEW.<col> is NULL and the trigger would write NULL
# into a possibly NOT NULL raw column -- defeating the column's
# DEFAULT. COALESCE onto the declared default so the raw column's
# server_default still applies when the caller doesn't set it.
col_default = literal_defaults(primary.columns)
new_parts = [coalesced_value(c, col_default) for c in insert_cols]
set_parts = [
f"{c} = {coalesced_value(c, col_default)}" for c in insert_cols
]
if "updated_at_column" in ctx:
set_parts.append(f"{ctx['updated_at_column']} = now()")
template_vars = {
"base_table": base_fullname,
"cols": ", ".join(insert_cols),
"new_cols": ", ".join(new_parts),
"set_clause": ", ".join(set_parts),
"returning_cols": _build_returning(primary, ctx, columns, dim_cols),
"pk_col": ctx.pk_column_name,
}
return [
TriggerOp(
"insert",
load_template(_TEMPLATES / "insert.plpgsql.mako").render(
**template_vars
),
),
TriggerOp(
"update",
load_template(_TEMPLATES / "update.plpgsql.mako").render(
**template_vars
),
),
render_delete_op(ctx, _TEMPLATES, template_vars),
]
return builder
[docs]
@produces("raw_table")
@requires(
"pk_columns",
"created_at_column",
"updated_at_column",
"exclude_from_mutations",
"deleted_at_column",
)
@singleton("__table__")
class SimpleTablePlugin(Plugin):
"""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
:func:`SimpleViewPlugin`.
"""
[docs]
def run(self, ctx: FactoryContext) -> None:
"""Create the raw table and store it in ctx."""
raw_name = resolve_name(
ctx.metadata,
"simple_raw_table",
{
"table_name": ctx.tablename,
"schema": ctx.schemaname,
},
_NAMING_DEFAULTS,
)
pk_columns = ctx["pk_columns"]
# Auto-managed timestamp columns get ``server_default=now()``
# plus ``NOT NULL`` so the view's ``INSTEAD OF INSERT``
# trigger can omit them from the column list and let
# PostgreSQL fill in the timestamp.
timestamp_cols: list[Column] = []
if "created_at_column" in ctx:
timestamp_cols.append(
Column(
ctx["created_at_column"],
DateTime(timezone=True),
server_default=func.now(),
nullable=False,
)
)
if "updated_at_column" in ctx:
timestamp_cols.append(
Column(
ctx["updated_at_column"],
DateTime(timezone=True),
server_default=func.now(),
nullable=False,
)
)
table = Table(
raw_name,
ctx.metadata,
*pk_columns,
*ctx.table_items,
*timestamp_cols,
*soft_delete_columns(ctx),
schema=ctx.schemaname,
)
ctx["raw_table"] = table
def _cloned_server_default(source: Column) -> FetchedValue | None:
"""Re-wrap a raw column's server default for the view proxy.
A :class:`~sqlalchemy.schema.DefaultClause` is bound to the column
it was declared on, so it can't be shared onto a second column --
re-wrap its arg in a fresh instance (same idiom as
:meth:`PrimaryKeyColumns.make_derived`). Carrying the default onto
the proxy is what tells SQLAlchemy the *database* supplies the value
on insert (the view's ``INSTEAD OF`` trigger routes to the raw table
where ``gen_random_uuid()`` / ``now()`` actually fire): it stops the
"primary key column has no default" warning when a write omits the
value and lets the id be read back via ``RETURNING``. Pure mapping
metadata -- the view owns its columns, so no DDL is emitted.
"""
sd = source.server_default
if isinstance(sd, DefaultClause):
return DefaultClause(sd.arg, for_update=sd.for_update)
return sd
[docs]
def SimpleViewPlugin() -> ViewPlugin: # noqa: N802
"""Create a configured ViewPlugin for simple dimensions.
Registers ``{tablename}`` as a view over ``{tablename}_raw``
and stores the proxy in ``ctx["primary"]``.
Returns:
A :class:`~codegen_database.plugins.view.ViewPlugin` configured
for simple passthrough views.
"""
def _query(ctx: FactoryContext) -> str:
raw = ctx["raw_table"]
cols: list[ColumnClause] = [column(c.name) for c in raw.columns]
return compile_query(select(*cols).select_from(raw))
def _proxy(ctx: FactoryContext) -> list[Column]:
raw = ctx["raw_table"]
return [
Column(
c.name,
c.type,
primary_key=c.primary_key,
server_default=_cloned_server_default(c),
)
for c in raw.columns
]
return ViewPlugin(
query_builder=_query,
proxy_builder=_proxy,
extra_requires=["raw_table", "pk_columns"],
)
[docs]
class CodegenDatabaseSimple(ResourceFactory):
"""Create a simple dimension: one table with optional checks.
Internal plugins (always present):
1. :class:`SimpleTablePlugin` -- raw backing table
(``{tablename}_raw``).
2. :class:`SimpleViewPlugin` -- writable view (``{tablename}``).
3. :class:`~codegen_database.plugins.check.TableCheckPlugin` --
check constraints on the raw table.
4. :class:`~codegen_database.plugins.index.TableIndexPlugin` --
indexes on the raw table.
5. :class:`~codegen_database.plugins.fk.TableFKPlugin` --
foreign keys on the raw table.
6. :class:`~codegen_database.plugins.protect.RawTableProtectionPlugin` --
blocks direct DML on the raw table.
7. :class:`~codegen_database.plugins.trigger.InsteadOfTriggerPlugin`
-- INSTEAD OF triggers on the dimension view.
A :class:`~codegen_database.plugins.pk.SerialPKPlugin` is auto-added
when no user plugin produces ``pk_columns``.
Args:
tablename: Name of the dimension table.
schemaname: PostgreSQL schema for generated objects.
metadata: SQLAlchemy ``MetaData`` to register on.
schema_items: Column and constraint definitions.
plugins: Behaviour-modifying plugins (e.g.
``UUIDV4PKPlugin``).
extra_plugins: Appended to the resolved plugin list.
"""
_FK_TARGET_KEY: ClassVar[str] = "raw_table"
_INTERNAL_PLUGINS: ClassVar[list[PluginOrCollection]] = [
SimpleTablePlugin(),
SimpleViewPlugin(),
TableCheckPlugin("raw_table"),
TableIndexPlugin("raw_table"),
TableFKPlugin("raw_table"),
RawTableProtectionPlugin("raw_table"),
InsteadOfTriggerPlugin(
ops_builder=_build_simple_ops_with_columns(None),
naming_defaults=_NAMING_DEFAULTS,
function_key="simple_function",
trigger_key="simple_trigger",
view_key="primary",
extra_requires=["raw_table"],
),
]