r"""Unified activity/audit query helpers.
Companion to :mod:`codegen_database.ext.audit.queries`. Where the
``construct_*_diff_query`` helpers emit the *raw* per-table transition
shape (wide ``<col>_before`` / ``<col>_after`` pairs for append-only,
long ``value_before`` / ``value_after`` for EAV), these helpers fold
both layouts into one uniform *activity* row::
(key..., changed_at, actor, field, before, after, change_type)
- ``changed_at`` -- when the transition happened.
- ``actor`` -- who made it. ``NULL`` for historical rows: today's
append-only / EAV backing tables carry only ``created_at``, no actor
column (Path A -- non-invasive). Forward-only capture plugs in here
once a resource records an actor.
- ``field`` -- the column (append-only) or attribute name (EAV) that
changed.
- ``before`` / ``after`` -- the prior and new values, cast to ``text``
for a uniform wire shape (the FE renders them as strings).
- ``change_type`` -- ``"create"`` for the first observation of a key
/ attribute (``before IS NULL``), ``"update"`` otherwise.
Both helpers return a SQLAlchemy :class:`~sqlalchemy.Select` without
registering anything on metadata. Execute directly against a
connection.
"""
from __future__ import annotations
from typing import TYPE_CHECKING, Any
from sqlalchemy import (
FromClause,
Text,
case,
func,
literal,
null,
select,
)
from sqlalchemy import (
cast as sa_cast,
)
from codegen_database.errors import CodegenDatabaseValidationError
if TYPE_CHECKING:
from sqlalchemy import CompoundSelect, Select
# Re-export the column validators so callers + tests reach them through
# one module without importing the queries module separately.
from codegen_database.ext.audit.queries import ( # noqa: F401
_resolve_eav_columns,
_validate_columns,
)
[docs]
def construct_append_only_activity_query(
table: FromClause,
*,
key_cols: list[str],
tracked_cols: list[str] | None = None,
ts_col: str = "created_at",
) -> CompoundSelect[Any]:
"""Unified activity stream over an append-only (SCD Type 2) table.
Wraps the same ``LAG()`` partitioned by *key_cols* that
:func:`~codegen_database.ext.audit.construct_append_only_diff_query`
uses, then *unpivots* the wide ``<col>_before`` / ``<col>_after``
pairs into one row per changed field -- so a transition that
touched three columns yields three activity rows (one per field)
rather than one wide row. Each row carries ``actor = NULL``
(Path A) and a ``change_type`` of ``"create"`` (first version of
the key, ``before IS NULL``) or ``"update"``.
Generates (conceptually)::
WITH lagged AS (
SELECT <key...>, <ts> AS changed_at,
lag(<col1>) OVER w AS <col1>_before, <col1> AS <col1>_after,
...
FROM <table>
WINDOW w AS (PARTITION BY <key...> ORDER BY <ts>)
)
SELECT <key...>, changed_at, NULL AS actor, '<col1>' AS field,
<col1>_before::text AS before, <col1>_after::text AS after,
CASE WHEN <col1>_before IS NULL THEN 'create' ELSE 'update' END
AS change_type
FROM lagged WHERE <col1>_before IS DISTINCT FROM <col1>_after
UNION ALL
SELECT ... '<col2>' ...
WHERE <col2>_before IS DISTINCT FROM <col2>_after
...
ORDER BY <key...>, changed_at, field
Args:
table: The append-only history table.
key_cols: Column names identifying the audited entity
(the ``LAG`` partition).
tracked_cols: Columns whose transitions to surface. Defaults
to every non-key, non-timestamp, non-PK column.
ts_col: Timestamp column used for ordering. Defaults to
``"created_at"``.
Returns:
A SQLAlchemy :class:`~sqlalchemy.Select` with columns
``(*key_cols, changed_at, actor, field, before, after,
change_type)``.
Raises:
CodegenDatabaseValidationError: If *key_cols* is empty or a
referenced column is missing.
"""
key, tracked, ts = _validate_columns(
table,
key_cols=key_cols,
ts_col=ts_col,
tracked_cols=tracked_cols,
)
# Build the lagged subquery once; each per-field select reads it.
lagged_cols: list[Any] = [k.label(k.key) for k in key]
lagged_cols.append(ts.label("changed_at"))
for c in tracked:
lagged_cols.append(
func.lag(c)
.over(partition_by=key, order_by=ts)
.label(f"{c.name}_before")
)
lagged_cols.append(c.label(f"{c.name}_after"))
lagged = select(*lagged_cols).select_from(table).subquery()
# One SELECT per tracked field, UNION ALL'd. Each filters to rows
# where *that* field changed, so a wide transition yields one row
# per changed field rather than one wide row.
parts: list[Select[Any]] = []
for c in tracked:
before = lagged.c[f"{c.name}_before"]
after = lagged.c[f"{c.name}_after"]
parts.append(
select(
*[lagged.c[k.key] for k in key],
lagged.c["changed_at"],
null().cast(Text).label("actor"),
literal(c.name, type_=Text).label("field"),
sa_cast(before, Text).label("before"),
sa_cast(after, Text).label("after"),
case(
(before.is_(None), "create"),
else_="update",
).label("change_type"),
).where(before.is_distinct_from(after))
)
if not parts:
msg = "no tracked columns to build an activity stream from"
raise CodegenDatabaseValidationError(msg)
return (
parts[0]
.union_all(*parts[1:])
.order_by(
*[parts[0].selected_columns[k.key] for k in key],
parts[0].selected_columns["changed_at"],
parts[0].selected_columns["field"],
)
)
[docs]
def construct_eav_activity_query( # noqa: PLR0913
table: FromClause,
*,
entity_col: str = "entity_id",
attribute_col: str = "attribute_name",
value_cols: list[str] | None = None,
attributes: list[str] | None = None,
ts_col: str = "created_at",
) -> Select[Any]:
"""Unified activity stream over an EAV attribute log.
The EAV diff is already long (one row per ``(entity, attribute)``
transition), so this wraps
:func:`~codegen_database.ext.audit.construct_eav_diff_query` and
adds the two missing unified columns: ``actor = NULL`` (Path A)
and ``change_type`` (``"create"`` for the first observation of an
attribute, ``"update"`` otherwise).
Args:
table: The EAV attribute log.
entity_col: Column naming the entity. Defaults to
``"entity_id"``.
attribute_col: Column naming the attribute. Defaults to
``"attribute_name"``.
value_cols: The typed value columns. Defaults to every
column ending in ``"_value"``.
attributes: Optional list of attribute names to restrict to.
ts_col: Timestamp column. Defaults to ``"created_at"``.
Returns:
A SQLAlchemy :class:`~sqlalchemy.Select` with columns
``(<entity_col>, <attribute_col>, changed_at, actor, field,
before, after, change_type)``. ``field`` mirrors the
attribute name so the column layout matches the append-only
helper.
Raises:
CodegenDatabaseValidationError: If a referenced column is
missing.
"""
from codegen_database.ext.audit.queries import ( # noqa: PLC0415 -- avoid cycle
construct_eav_diff_query,
)
diff = construct_eav_diff_query(
table,
entity_col=entity_col,
attribute_col=attribute_col,
value_cols=value_cols,
attributes=attributes,
ts_col=ts_col,
).subquery()
# ``field`` = the attribute name, so both helpers expose the same
# column layout (the FE keys off ``field`` / ``before`` / ``after``
# without caring which layout backed the resource).
return select(
diff.c[entity_col],
diff.c[attribute_col],
diff.c["changed_at"],
null().cast(Text).label("actor"),
diff.c[attribute_col].label("field"),
diff.c["value_before"].label("before"),
diff.c["value_after"].label("after"),
case(
(diff.c["value_before"].is_(None), "create"),
else_="update",
).label("change_type"),
).order_by(
diff.c[entity_col],
diff.c[attribute_col],
diff.c["changed_at"],
)
__all__ = [
"construct_append_only_activity_query",
"construct_eav_activity_query",
]