r"""Changeset activity view plugin (append-only + EAV).
Produces a ``{table_name}_activity`` SQL view that folds per-field
transition rows into one row per *changeset* (version), with a JSONB
``fields`` array of ``{field, before, after}`` objects.
Both append-only and EAV backing tables have a ``BigInteger`` surrogate
``id`` PK (unique per version/observation) and an entity-linking column
(``id_entity`` for append-only, ``entity_id`` for EAV). The plugin
reads whichever is present and produces the same unified view shape::
id (BigInteger, the backing table's surrogate PK),
entity_id (the dimension's PK type),
changed_at (timestamptz),
actor (text, NULL for historical rows),
change_type ('create' | 'update'),
fields (jsonb: {changes: [{field, before, after}, ...]})
Auto-included by append-only + EAV factories
and :class:`~codegen_database.factory.CodegenDatabaseEAV` -- no
consumer opt-in needed. Point a codegen-be resource at the view
(``CodegenDatabaseViewMixin`` subclass) to expose list / get / SDK.
"""
from __future__ import annotations
from typing import TYPE_CHECKING, Any
from sqlalchemy import (
BigInteger,
Column,
DateTime,
String,
Table,
Text,
func,
select,
)
from sqlalchemy.dialects.postgresql import JSONB
from sqlalchemy_declarative_extensions import View, register_view
if TYPE_CHECKING:
from codegen_database.factory.context import FactoryContext
from codegen_database.ext.audit import (
construct_append_only_activity_query,
construct_eav_activity_query,
)
from codegen_database.plugin import Plugin, produces, requires
from codegen_database.utils.naming import resolve_name
from codegen_database.utils.query import compile_query
_NAMING_DEFAULTS = {
"activity_view": "%(table_name)s_activity",
}
[docs]
@produces("activity_view")
@requires("pk_columns", "created_at_column")
class ActivityViewPlugin(Plugin):
"""Generate a ``{table_name}_activity`` changeset view.
Auto-included by both append-only and EAV factories. Detects
which backing table is present (``attributes`` for append-only,
``attribute`` for EAV) and builds the activity view accordingly.
The view is keyed by the backing table's ``BigInteger`` surrogate
``id`` so keyset pagination works -- the entity UUID repeats
across versions and can't page.
"""
[docs]
def run(self, ctx: FactoryContext) -> None:
"""Create the activity view on the shared metadata."""
attr_table = _resolve_attributes_table(ctx)
query = _build_changeset_query(ctx, attr_table)
view_name = resolve_name(
ctx.metadata,
"activity_view",
{"table_name": ctx.tablename, "schema": ctx.schemaname},
_NAMING_DEFAULTS,
)
compiled = compile_query(query)
# Register the view via sqlalchemy_declarative_extensions so
# alembic renders CREATE VIEW (not CREATE TABLE). No Table
# proxy on ctx.metadata -- ``register_view`` handles DDL;
# a proxy would trigger ``create_all`` and conflict.
register_view(
ctx.metadata,
View(
view_name,
compiled,
schema=ctx.schemaname,
),
)
# Build a detached Table proxy (separate MetaData) for
# downstream consumers that need column introspection,
# without registering it for DDL emission.
from sqlalchemy import MetaData # noqa: PLC0415
proxy = Table(
view_name,
MetaData(),
Column("id", BigInteger, primary_key=True),
Column("entity_id", ctx["pk_columns"].first.type),
Column("changed_at", DateTime(timezone=True)),
Column("actor", Text),
Column("change_type", String),
Column("fields", JSONB),
schema=ctx.schemaname,
)
ctx["activity_view"] = proxy
def _resolve_attributes_table(ctx: FactoryContext) -> Table:
"""Find the backing history table (append-only or EAV)."""
for key in ("attributes", "attribute"):
table = ctx.get(key)
if table is not None:
return table
msg = (
"ActivityViewPlugin: no 'attributes' or 'attribute' table "
"in ctx -- is this an append-only or EAV dimension?"
)
raise RuntimeError(msg)
def _resolve_entity_col(ctx: FactoryContext, table: Table) -> str:
"""Find the entity-linking column on the backing table."""
pk_col_name = ctx.pk_column_name
# Append-only stamps ``id_entity``; EAV uses ``entity_id``.
for candidate in (f"{pk_col_name}_entity", "entity_id"):
if candidate in table.c:
return candidate
msg = (
f"ActivityViewPlugin: no entity column on {table.name!r} "
f"(looked for '{pk_col_name}_entity' and 'entity_id')"
)
raise RuntimeError(msg)
def _build_changeset_query(
ctx: FactoryContext,
attr_table: Table,
) -> Any: # noqa: ANN401
"""Build the GROUP BY changeset view query.
Dispatches to the append-only or EAV activity helper depending on
the table shape, then folds per-field rows into changesets.
"""
pk_col_name = ctx.pk_column_name
entity_col = _resolve_entity_col(ctx, attr_table)
is_eav = "attribute_name" in attr_table.c
if is_eav:
return _build_eav_changeset_query(attr_table, entity_col)
return _build_append_only_changeset_query(
attr_table, entity_col, pk_col_name
)
def _build_append_only_changeset_query(
attr_table: Table,
entity_col: str,
pk_col_name: str, # noqa: ARG001 -- reserved for future use
) -> Any: # noqa: ANN401
"""Append-only: one row per version, fields grouped by json_agg."""
tracked = [
c
for c in attr_table.c
if c.name not in {"id", "created_at", "deleted_at", entity_col}
]
history_cols: list[Any] = [
attr_table.c.id.label("id"),
attr_table.c[entity_col].label("entity_id"),
attr_table.c.created_at.label("created_at"),
*[c.label(c.key) for c in tracked],
]
history = select(*history_cols).select_from(attr_table).subquery()
activity = construct_append_only_activity_query(
history,
key_cols=["entity_id"],
tracked_cols=[c.key for c in tracked],
ts_col="created_at",
).subquery()
activity_full = (
select(
history.c.id.label("id"),
activity.c.entity_id.label("entity_id"),
activity.c.changed_at.label("changed_at"),
activity.c.actor.label("actor"),
activity.c.change_type.label("change_type"),
activity.c.field.label("field"),
activity.c.before.label("before"),
activity.c.after.label("after"),
)
.select_from(
activity.join(
history,
(activity.c.entity_id == history.c.entity_id)
& (activity.c.changed_at == history.c.created_at),
),
)
.subquery()
)
return (
select(
activity_full.c.id.label("id"),
activity_full.c.entity_id.label("entity_id"),
activity_full.c.changed_at.label("changed_at"),
activity_full.c.actor.label("actor"),
activity_full.c.change_type.label("change_type"),
func.json_build_object(
"changes",
func.json_agg(
func.json_build_object(
"field",
activity_full.c.field,
"before",
activity_full.c.before,
"after",
activity_full.c.after,
),
),
).label("fields"),
)
.group_by(
activity_full.c.id,
activity_full.c.entity_id,
activity_full.c.changed_at,
activity_full.c.actor,
activity_full.c.change_type,
)
.order_by(activity_full.c.changed_at.desc())
)
def _build_eav_changeset_query(
attr_table: Table,
entity_col: str,
) -> Any: # noqa: ANN401
"""EAV: one row per (entity, timestamp) changeset, fields grouped.
EAV attribute rows are already one-per-observation. Group by
``(entity_id, created_at, change_type)`` so all attributes changed
at the same timestamp (e.g. a multi-field update) fold into one
changeset.
"""
activity = construct_eav_activity_query(
attr_table,
entity_col=entity_col,
).subquery()
# EAV activity rows are per-attribute; group by (entity, timestamp,
# change_type) so multi-field changes at the same timestamp fold
# into one changeset. Use a row_number() window for a stable
# unique id per changeset (keyset pagination needs a unique row id).
grouped = select(
func.dense_rank()
.over(
order_by=[
activity.c[entity_col],
activity.c.changed_at,
activity.c.change_type,
]
)
.label("id"),
activity.c[entity_col].label("entity_id"),
activity.c.changed_at.label("changed_at"),
activity.c.actor.label("actor"),
activity.c.change_type.label("change_type"),
activity.c.field.label("field"),
activity.c.before.label("before"),
activity.c.after.label("after"),
).subquery()
return (
select(
grouped.c.id.label("id"),
grouped.c.entity_id.label("entity_id"),
grouped.c.changed_at.label("changed_at"),
grouped.c.actor.label("actor"),
grouped.c.change_type.label("change_type"),
func.json_build_object(
"changes",
func.json_agg(
func.json_build_object(
"field",
grouped.c.field,
"before",
grouped.c.before,
"after",
grouped.c.after,
),
),
).label("fields"),
)
.group_by(
grouped.c.id,
grouped.c.entity_id,
grouped.c.changed_at,
grouped.c.actor,
grouped.c.change_type,
)
.order_by(grouped.c.changed_at.desc())
)
__all__ = ["ActivityViewPlugin"]