"""Ledger resource factory."""
from __future__ import annotations
from pathlib import Path
from typing import TYPE_CHECKING, ClassVar
from sqlalchemy import Column, DateTime, Table, func, select, text
from sqlalchemy.dialects.postgresql import UUID
from codegen_database.factory.base import ResourceFactory
from codegen_database.plugin import Plugin, produces, requires, singleton
from codegen_database.plugins.check import TableCheckPlugin
from codegen_database.plugins.column_name import construct_column_name_plugin
from codegen_database.plugins.fk import TableFKPlugin
from codegen_database.plugins.index import TableIndexPlugin
from codegen_database.plugins.ledger import (
_NAMING_DEFAULTS as _LEDGER_NAMING,
)
from codegen_database.plugins.protect import RawTableProtectionPlugin
from codegen_database.plugins.trigger import InsteadOfTriggerPlugin, TriggerOp
from codegen_database.plugins.view import ViewPlugin
from codegen_database.types import build_value_type
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 import MetaData
from sqlalchemy.schema import SchemaItem
from codegen_database.check import CodegenDatabaseCheck
from codegen_database.factory.context import FactoryContext
from codegen_database.index import CodegenDatabaseIndex
from codegen_database.plugin import PluginOrCollection
from codegen_database.types import ValueType
_TEMPLATES = (
Path(__file__).resolve().parents[1] / "plugins" / "templates" / "ledger"
)
_FACTORY_NAMING_DEFAULTS = {
"ledger_raw_table": "%(table_name)s_raw",
}
[docs]
@produces("entry_id_column")
@singleton("__entry_id__")
class UUIDEntryIDPlugin(Plugin):
"""Provide a UUIDv4 entry ID column for ledger tables.
Stores a :class:`~sqlalchemy.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.
Args:
column_name: Name of the entry ID column
(default ``"entry_id"``).
"""
def __init__(self, column_name: str = "entry_id") -> None:
"""Store the column name."""
self._column_name = column_name
[docs]
def run(self, ctx: FactoryContext) -> None:
"""Store the entry ID column and inject it."""
col = Column(
self._column_name,
UUID(as_uuid=True),
nullable=False,
server_default=text("gen_random_uuid()"),
)
ctx["entry_id_column"] = col
ctx.injected_columns.append(col)
[docs]
@produces("raw_table")
@requires("pk_columns", "entry_id_column", "created_at_column")
@singleton("__table__")
class LedgerTablePlugin(Plugin):
"""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
:func:`LedgerViewPlugin`.
Args:
value_type: Type for the value column. One of
``"integer"``, ``"numeric"``, or ``"decimal"``
(default ``"integer"``).
precision: Total number of digits when *value_type* is
``"numeric"`` / ``"decimal"`` -- the
``NUMERIC(precision, scale)`` first argument. ``None``
leaves the column unconstrained.
scale: Digits after the decimal point. Requires
*precision*.
Raises:
CodegenDatabaseValidationError: If *value_type* is not a
recognised type, or precision/scale are misconfigured.
"""
def __init__(
self,
value_type: ValueType = "integer",
*,
precision: int | None = None,
scale: int | None = None,
) -> None:
"""Store configuration."""
# Validate eagerly so a bad type / precision combo fails at
# construction; the column itself is built fresh in ``run``.
build_value_type(value_type, precision=precision, scale=scale)
self.value_type = value_type
self.precision = precision
self.scale = scale
[docs]
def run(self, ctx: FactoryContext) -> None:
"""Create the ledger raw table and store it in ctx."""
pk_columns = ctx["pk_columns"]
created_at_col = ctx["created_at_column"]
value_col_type = build_value_type(
self.value_type,
precision=self.precision,
scale=self.scale,
)
raw_name = resolve_name(
ctx.metadata,
"ledger_raw_table",
{
"table_name": ctx.tablename,
"schema": ctx.schemaname,
},
_FACTORY_NAMING_DEFAULTS,
)
table = Table(
raw_name,
ctx.metadata,
*pk_columns,
*ctx.injected_columns,
Column(
created_at_col,
DateTime(timezone=True),
server_default=func.now(),
),
Column("value", value_col_type, nullable=False),
*ctx.table_items,
schema=ctx.schemaname,
)
ctx["raw_table"] = table
[docs]
def LedgerViewPlugin() -> ViewPlugin: # noqa: N802
"""Create a configured ViewPlugin for ledger 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 ledger passthrough views.
"""
def _query(ctx: FactoryContext) -> str:
raw = ctx["raw_table"]
return compile_query(select(raw))
def _proxy(ctx: FactoryContext) -> list[Column]:
raw = ctx["raw_table"]
return [
Column(c.name, c.type, primary_key=c.primary_key)
for c in raw.columns
]
return ViewPlugin(
query_builder=_query,
proxy_builder=_proxy,
extra_requires=[
"raw_table",
"pk_columns",
"entry_id_column",
"created_at_column",
],
)
def _make_ledger_ops_builder(
table_key: str,
view_key: str,
) -> Callable[[FactoryContext], list[TriggerOp]]:
"""Return an ops builder for ledger dimensions."""
def build(ctx: FactoryContext) -> list[TriggerOp]:
raw_table = ctx[table_key]
base_fullname = f"{ctx.schemaname}.{raw_table.name}"
entry_id_col = ctx["entry_id_column"]
dim_cols = ctx.dim_column_names
# Include injected columns that have no server default and are not
# entry_id (already handled). This picks up e.g. the `direction`
# column added by DoubleEntryPlugin which must be provided by the
# caller — unlike `created_at` which has a server default.
injected_writable = [
col.key
for col in ctx.injected_columns
if col.key != entry_id_col.name and col.server_default is None
]
all_cols = [
entry_id_col.name,
"value",
*dim_cols,
*injected_writable,
]
new_col_exprs = []
for c in all_cols:
if c == entry_id_col.name:
default = entry_id_col.server_default.arg.text
new_col_exprs.append(f"COALESCE(NEW.{c}, {default})")
else:
new_col_exprs.append(f"NEW.{c}")
view_proxy = ctx[view_key]
view_schema = view_proxy.schema or ctx.schemaname
view_fullname = f"{view_schema}.{ctx.tablename}"
template_vars = {
"base_table": base_fullname,
"cols": ", ".join(all_cols),
"new_cols": ", ".join(new_col_exprs),
"view": view_fullname,
}
return [
TriggerOp(
"insert",
load_template(_TEMPLATES / "insert.plpgsql.mako").render(
**template_vars
),
),
TriggerOp(
"update",
load_template(_TEMPLATES / "reject_update.plpgsql.mako").render(
**template_vars
),
),
TriggerOp(
"delete",
load_template(_TEMPLATES / "reject_delete.plpgsql.mako").render(
**template_vars
),
),
]
return build
[docs]
class CodegenDatabaseLedger(ResourceFactory):
"""Create a ledger: append-only table with a value column.
Internal plugins (always present), in order:
1. :class:`UUIDEntryIDPlugin` -- UUID entry ID for correlating
related entries.
Next, the column-name plugin
:func:`~codegen_database.plugins.column_name.construct_column_name_plugin`
sets the ``created_at`` column name. Then:
2. :class:`LedgerTablePlugin` -- raw backing table
(``{tablename}_raw``).
3. :class:`LedgerViewPlugin` -- writable view (``{tablename}``).
4. :class:`~codegen_database.plugins.protect.RawTableProtectionPlugin` --
blocks direct DML on the raw table.
5. :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``.
Use
:func:`~codegen_database.ext.ledger.queries.construct_ledger_balance_query`,
:func:`~codegen_database.ext.ledger.queries.construct_ledger_latest_query`,
and
:func:`~codegen_database.ext.ledger.functions.ledger_event_function` with
:class:`~codegen_database.functions.CodegenDatabaseFunction` for derived
views and event functions.
"""
_FK_TARGET_KEY: ClassVar[str] = "raw_table"
_INTERNAL_PLUGINS: ClassVar[list[PluginOrCollection]] = [
UUIDEntryIDPlugin(),
construct_column_name_plugin("created_at_column", "created_at"),
LedgerTablePlugin(),
LedgerViewPlugin(),
TableCheckPlugin("raw_table"),
TableIndexPlugin("raw_table"),
TableFKPlugin("raw_table"),
RawTableProtectionPlugin("raw_table"),
InsteadOfTriggerPlugin(
ops_builder=_make_ledger_ops_builder("raw_table", "primary"),
naming_defaults=_LEDGER_NAMING,
function_key="ledger_function",
trigger_key="ledger_trigger",
view_key="primary",
extra_requires=[
"raw_table",
"entry_id_column",
],
),
]
def __init__( # noqa: PLR0913
self,
tablename: str,
schemaname: str,
metadata: MetaData,
schema_items: list[
SchemaItem | CodegenDatabaseCheck | CodegenDatabaseIndex
],
*,
config: object | None = None,
plugins: list[PluginOrCollection] | None = None,
extra_plugins: list[PluginOrCollection] | None = None,
) -> None:
"""Create the ledger and register it on *metadata*.
Args:
tablename: Name of the ledger table.
schemaname: PostgreSQL schema for generated objects.
metadata: SQLAlchemy ``MetaData`` to register on.
schema_items: Dimension column definitions.
config: Optional global config.
plugins: If given, replaces ``DEFAULT_PLUGINS``.
extra_plugins: Appended to resolved plugin list.
"""
super().__init__(
tablename,
schemaname,
metadata,
schema_items,
config=config,
plugins=plugins,
extra_plugins=extra_plugins,
)