"""Query builders for history/audit tables.
Pure-Python helpers that return SQLAlchemy selects. Two dimension
layouts are supported:
- **Append-only (SCD Type 2):** each version is a full row, keyed by
the dimension's business columns. Use
:func:`construct_append_only_diff_query` and
:func:`construct_append_only_changed_between_query`. For a
:class:`~codegen_database.factory.dimension.append_only.CodegenDatabaseAppendOnly`
factory, pass ``factory.ctx["attributes"]``.
- **EAV (Entity-Attribute-Value):** each row is a single attribute's
value at a point in time. Use :func:`construct_eav_diff_query` and
:func:`construct_eav_changed_between_query`. For a
:class:`~codegen_database.factory.dimension.eav.CodegenDatabaseEAV`
factory, pass ``factory.ctx["attribute"]``.
All helpers emit typed ``<col>_before`` / ``<col>_after`` (or
``value_before`` / ``value_after`` for EAV) pairs. No JSONB.
"""
from __future__ import annotations
from typing import TYPE_CHECKING, Any
from sqlalchemy import Column, FromClause, Text, and_, func, or_, select
from sqlalchemy import cast as sa_cast
from codegen_database.errors import CodegenDatabaseValidationError
if TYPE_CHECKING:
from datetime import datetime
from sqlalchemy import Select
from sqlalchemy.sql import ColumnElement, Subquery
_TimeBound = datetime | ColumnElement[Any]
def _pk_names(table: FromClause) -> set[str]:
"""Primary-key column names of *table*, or empty for a join/subquery.
``Table`` carries a primary-key constraint; a ``Join`` / subquery
exposes an empty ``ColumnSet`` with no ``.columns`` attribute.
Tolerate both so the helpers accept a pre-joined selectable
(e.g. an append-only ``attributes JOIN root`` that exposes the
entity id the raw ``attributes`` table lacks).
"""
pk = getattr(table, "primary_key", None)
if pk is None:
return set()
pk_cols = getattr(pk, "columns", None)
if pk_cols is None:
return set()
return {c.name for c in pk_cols}
def _validate_columns(
table: FromClause,
*,
key_cols: list[str],
ts_col: str,
tracked_cols: list[str] | None,
) -> tuple[list[Any], list[Any], Any]:
"""Validate column names and resolve them against *table*.
Returns the resolved ``(key, tracked, ts)`` columns. The default
``tracked`` is every column not in *key_cols*, not the timestamp,
and not part of the primary key -- surrogate version IDs should
not appear in diff output.
"""
if not key_cols:
msg = "key_cols must be a non-empty list"
raise CodegenDatabaseValidationError(msg)
available = set(table.c.keys())
table_label = getattr(table, "name", "<selectable>")
missing_keys = [c for c in key_cols if c not in available]
if missing_keys:
msg = f"key_cols not in table {table_label!r}: {missing_keys}"
raise CodegenDatabaseValidationError(msg)
if ts_col not in available:
msg = f"ts_col {ts_col!r} not in table {table_label!r}"
raise CodegenDatabaseValidationError(msg)
if ts_col in key_cols:
msg = "ts_col must not appear in key_cols"
raise CodegenDatabaseValidationError(msg)
pk_names = _pk_names(table)
if tracked_cols is None:
exclude = set(key_cols) | {ts_col} | pk_names
tracked = [c for c in table.c if c.name not in exclude]
else:
missing_tracked = [c for c in tracked_cols if c not in available]
if missing_tracked:
msg = (
f"tracked_cols not in table {table_label!r}: {missing_tracked}"
)
raise CodegenDatabaseValidationError(msg)
overlap = set(tracked_cols) & (set(key_cols) | {ts_col})
if overlap:
msg = (
f"tracked_cols overlap with key_cols/ts_col: {sorted(overlap)}"
)
raise CodegenDatabaseValidationError(msg)
tracked = [table.c[n] for n in tracked_cols]
if not tracked:
msg = (
"tracked_cols is empty -- no columns to diff. "
"Either add columns to the table or pass tracked_cols "
"explicitly."
)
raise CodegenDatabaseValidationError(msg)
key = [table.c[n] for n in key_cols]
ts = table.c[ts_col]
return key, tracked, ts
# -- append-only (SCD Type 2) -------------------------------------------
[docs]
def construct_append_only_diff_query(
table: FromClause,
*,
key_cols: list[str],
tracked_cols: list[str] | None = None,
ts_col: str = "created_at",
) -> Select[Any]:
"""Per-transition diff over an append-only (SCD Type 2) history table.
Works on any table where each row represents a full version of an
entity -- typically the ``attributes`` backing table of a
:class:`~codegen_database.factory.dimension.append_only.CodegenDatabaseAppendOnly`
factory (``factory.ctx["attributes"]``), or a hand-rolled append-only
log. See :func:`construct_eav_diff_query` for the EAV equivalent.
Uses ``LAG()`` partitioned by *key_cols* ordered by *ts_col* to
pair each row with its predecessor. Emits one row per transition
where at least one tracked column changed (``IS DISTINCT FROM``).
The first row per key is included as a creation event: its
``*_before`` columns are ``NULL`` and ``*_after`` hold the initial
values.
Generates (conceptually)::
SELECT <key...>,
<ts> AS changed_at,
LAG(<col1>) OVER w AS <col1>_before, <col1> AS <col1>_after,
LAG(<col2>) OVER w AS <col2>_before, <col2> AS <col2>_after,
...
FROM <table>
WINDOW w AS (PARTITION BY <key...> ORDER BY <ts>)
WHERE <col1>_before IS DISTINCT FROM <col1>_after
OR <col2>_before IS DISTINCT FROM <col2>_after
OR ...
ORDER BY <key...>, changed_at
Args:
table: The append-only history table.
key_cols: Column names identifying the audited entity
(the partition for ``LAG``). Must be non-empty.
tracked_cols: Columns whose transitions to surface.
Defaults to every column that is not in *key_cols*,
not the timestamp, and not part of the primary key.
ts_col: Timestamp column used for ordering. Defaults to
``"created_at"``.
Returns:
A SQLAlchemy :class:`~sqlalchemy.Select` with columns
``(*key_cols, changed_at, <col>_before, <col>_after, ...)``.
Raises:
CodegenDatabaseValidationError: If *key_cols* is empty, a referenced
column is missing, or *tracked_cols* resolves to empty.
"""
key, tracked, ts = _validate_columns(
table,
key_cols=key_cols,
ts_col=ts_col,
tracked_cols=tracked_cols,
)
lagged_cols: list[Any] = [k.label(k.key) for k in key]
lagged_cols.append(ts.label("changed_at"))
for c in tracked:
lag_expr = (
func.lag(c)
.over(partition_by=key, order_by=ts)
.label(f"{c.name}_before")
)
lagged_cols.append(lag_expr)
lagged_cols.append(c.label(f"{c.name}_after"))
lagged = select(*lagged_cols).select_from(table).subquery()
change_conditions = [
lagged.c[f"{c.name}_before"].is_distinct_from(
lagged.c[f"{c.name}_after"]
)
for c in tracked
]
final_cols: list[Any] = [lagged.c[k.key] for k in key]
final_cols.append(lagged.c["changed_at"])
for c in tracked:
final_cols.append(lagged.c[f"{c.name}_before"])
final_cols.append(lagged.c[f"{c.name}_after"])
return (
select(*final_cols)
.select_from(lagged)
.where(or_(*change_conditions))
.order_by(
*[lagged.c[k.key] for k in key],
lagged.c["changed_at"],
)
)
[docs]
def construct_append_only_changed_between_query( # noqa: PLR0913
table: FromClause,
*,
key_cols: list[str],
start: _TimeBound,
end: _TimeBound,
tracked_cols: list[str] | None = None,
ts_col: str = "created_at",
) -> Select[Any]:
"""Net-change query over a window on an append-only history table.
Companion to :func:`construct_append_only_diff_query`. For each key,
compares state-as-of *start* against state-as-of *end* (both
inclusive). Intermediate versions inside the window are collapsed
away -- only the net delta is returned. One row per key that
changed (including keys created or last-seen inside the window,
which appear with ``NULL`` on the missing side). See
:func:`construct_eav_changed_between_query` for the EAV equivalent.
Generates (conceptually)::
WITH
before_ AS (
SELECT DISTINCT ON (<key...>) <key...>, <col...>
FROM <table> WHERE <ts> <= :start
ORDER BY <key...>, <ts> DESC
),
after_ AS (
SELECT DISTINCT ON (<key...>) <key...>, <col...>
FROM <table> WHERE <ts> <= :end
ORDER BY <key...>, <ts> DESC
)
SELECT COALESCE(a.<key>, b.<key>) AS <key>, ...,
b.<col> AS <col>_before, a.<col> AS <col>_after, ...
FROM after_ a FULL JOIN before_ b USING (<key...>)
WHERE b.<col> IS DISTINCT FROM a.<col> OR ...
Args:
table: The append-only history table.
key_cols: Column names identifying the audited entity.
start: Lower bound of the window (inclusive). A
:class:`~datetime.datetime` or SQL expression.
end: Upper bound of the window (inclusive).
tracked_cols: Columns whose net change to surface. Defaults
to every non-key, non-timestamp, non-PK column.
ts_col: Timestamp column. Defaults to ``"created_at"``.
Returns:
A SQLAlchemy :class:`~sqlalchemy.Select` with columns
``(*key_cols, <col>_before, <col>_after, ...)``.
Raises:
CodegenDatabaseValidationError: If *key_cols* is empty, a referenced
column is missing, or *tracked_cols* resolves to empty.
"""
key, tracked, ts = _validate_columns(
table,
key_cols=key_cols,
ts_col=ts_col,
tracked_cols=tracked_cols,
)
def snapshot_at(bound: _TimeBound, name: str) -> Subquery:
cols: list[Any] = [k.label(k.key) for k in key]
cols.extend(c.label(c.key) for c in tracked)
return (
select(*cols)
.distinct(*key)
.where(ts <= bound)
.order_by(*key, ts.desc())
.subquery(name)
)
before = snapshot_at(start, "before_")
after = snapshot_at(end, "after_")
join_on = and_(*[before.c[k.key] == after.c[k.key] for k in key])
joined = before.join(after, join_on, full=True)
final_cols: list[Any] = [
func.coalesce(after.c[k.key], before.c[k.key]).label(k.key) for k in key
]
for c in tracked:
final_cols.append(before.c[c.name].label(f"{c.name}_before"))
final_cols.append(after.c[c.name].label(f"{c.name}_after"))
change_conditions = [
before.c[c.name].is_distinct_from(after.c[c.name]) for c in tracked
]
return (
select(*final_cols)
.select_from(joined)
.where(or_(*change_conditions))
.order_by(
*[func.coalesce(after.c[k.key], before.c[k.key]) for k in key]
)
)
# -- EAV ----------------------------------------------------------------
def _resolve_eav_columns(
table: FromClause,
*,
entity_col: str,
attribute_col: str,
value_cols: list[str] | None,
ts_col: str,
) -> tuple[Any, Any, list[Any], Any]:
"""Validate and resolve EAV columns against *table*.
Auto-detects *value_cols* as every column whose name ends with
``_value`` when not supplied. This matches the codegen_database EAV
convention (``integer_value``, ``text_value``, ...).
"""
available = set(table.c.keys())
table_label = getattr(table, "name", "<selectable>")
for name, role in (
(entity_col, "entity_col"),
(attribute_col, "attribute_col"),
(ts_col, "ts_col"),
):
if name not in available:
msg = f"{role} {name!r} not in table {table_label!r}"
raise CodegenDatabaseValidationError(msg)
if value_cols is None:
resolved_value_cols = [c for c in table.c if c.name.endswith("_value")]
if not resolved_value_cols:
msg = (
f"No columns ending in '_value' found on table "
f"{table_label!r}; pass value_cols explicitly."
)
raise CodegenDatabaseValidationError(msg)
else:
missing = [c for c in value_cols if c not in available]
if missing:
msg = f"value_cols not in table {table_label!r}: {missing}"
raise CodegenDatabaseValidationError(msg)
resolved_value_cols = [table.c[n] for n in value_cols]
return (
table.c[entity_col],
table.c[attribute_col],
resolved_value_cols,
table.c[ts_col],
)
def _coalesced_value(
value_cols: list[Column[Any]],
) -> ColumnElement[str]:
"""Coalesce each value column cast to text.
EAV rows carry exactly one non-null value column (enforced by the
factory's check constraint), so ``COALESCE`` collapses them into
a single typed expression without loss. Casting to ``text`` gives
a uniform output type -- useful for display diffs where the
caller does not want to care which typed column held the value.
"""
return func.coalesce(*[sa_cast(c, Text) for c in value_cols])
[docs]
def construct_eav_diff_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]:
"""Per-attribute diff over an EAV attribute log.
Works on any table shaped like the codegen_database EAV ``attribute`` table:
one row per (entity, attribute) value observation. For a
:class:`~codegen_database.factory.dimension.eav.CodegenDatabaseEAV`
factory, pass ``factory.ctx["attribute"]``. See
:func:`construct_append_only_diff_query`
for the append-only equivalent.
Uses ``LAG()`` partitioned by ``(entity, attribute)`` ordered by
*ts_col*. The first observation per ``(entity, attribute)``
comes through with ``value_before = NULL`` (a "first seen" event).
All value columns are coalesced into a single ``text``-typed
expression. EAV rows carry exactly one non-null value, so the
coalesce is loss-free modulo formatting.
Generates (conceptually)::
SELECT entity, attribute, <ts> AS changed_at,
LAG(value) OVER w AS value_before,
value AS value_after
FROM (
SELECT <entity>, <attribute>, <ts>,
COALESCE(<v1>::text, <v2>::text, ...) AS value
FROM <table>
[WHERE <attribute> IN :attributes]
) t
WINDOW w AS (PARTITION BY entity, attribute ORDER BY <ts>)
WHERE value_before IS DISTINCT FROM value_after
ORDER BY entity, attribute, changed_at
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
the diff to. ``None`` means all attributes.
ts_col: Timestamp column. Defaults to ``"created_at"``.
Returns:
A SQLAlchemy :class:`~sqlalchemy.Select` with columns
``(<entity_col>, <attribute_col>, changed_at, value_before,
value_after)``.
Raises:
CodegenDatabaseValidationError: If a referenced column is missing or
``value_cols`` resolves to empty.
"""
entity, attribute, values, ts = _resolve_eav_columns(
table,
entity_col=entity_col,
attribute_col=attribute_col,
value_cols=value_cols,
ts_col=ts_col,
)
value_expr = _coalesced_value(values).label("value")
base = select(
entity.label(entity.key),
attribute.label(attribute.key),
ts.label("changed_at"),
value_expr,
).select_from(table)
if attributes is not None:
base = base.where(attribute.in_(attributes))
base_sq = base.subquery("eav_rows")
lagged = select(
base_sq.c[entity.key].label(entity.key),
base_sq.c[attribute.key].label(attribute.key),
base_sq.c.changed_at.label("changed_at"),
func.lag(base_sq.c.value)
.over(
partition_by=[base_sq.c[entity.key], base_sq.c[attribute.key]],
order_by=base_sq.c.changed_at,
)
.label("value_before"),
base_sq.c.value.label("value_after"),
).subquery("lagged")
return (
select(
lagged.c[entity.key],
lagged.c[attribute.key],
lagged.c.changed_at,
lagged.c.value_before,
lagged.c.value_after,
)
.select_from(lagged)
.where(lagged.c.value_before.is_distinct_from(lagged.c.value_after))
.order_by(
lagged.c[entity.key],
lagged.c[attribute.key],
lagged.c.changed_at,
)
)
[docs]
def construct_eav_changed_between_query( # noqa: PLR0913
table: FromClause,
*,
start: _TimeBound,
end: _TimeBound,
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]:
"""Net-change query over a window on an EAV attribute log.
Companion to :func:`construct_eav_diff_query`. For each ``(entity,
attribute)`` pair, compares value-as-of *start* against value-as-of
*end* (both inclusive) via two ``DISTINCT ON`` snapshots joined
with a ``FULL JOIN``. One row per pair whose value differs; pairs
introduced inside the window appear with ``value_before = NULL``.
Args:
table: The EAV attribute log.
start: Lower bound of the window (inclusive).
end: Upper bound of the window (inclusive).
entity_col: Column naming the entity.
attribute_col: Column naming the attribute.
value_cols: Typed value columns; defaults to ``*_value``.
attributes: Optional attribute-name filter.
ts_col: Timestamp column.
Returns:
A SQLAlchemy :class:`~sqlalchemy.Select` with columns
``(<entity_col>, <attribute_col>, value_before, value_after)``.
Raises:
CodegenDatabaseValidationError: If a referenced column is missing or
``value_cols`` resolves to empty.
"""
entity, attribute, values, ts = _resolve_eav_columns(
table,
entity_col=entity_col,
attribute_col=attribute_col,
value_cols=value_cols,
ts_col=ts_col,
)
def snapshot_at(bound: _TimeBound, name: str) -> Subquery:
value_expr = _coalesced_value(values).label("value")
q = (
select(
entity.label(entity.key),
attribute.label(attribute.key),
value_expr,
)
.distinct(entity, attribute)
.where(ts <= bound)
.order_by(entity, attribute, ts.desc())
)
if attributes is not None:
q = q.where(attribute.in_(attributes))
return q.subquery(name)
before = snapshot_at(start, "before_")
after = snapshot_at(end, "after_")
# Use literal column names directly -- both subqueries share the
# same shape, so joining on entity/attribute keys is unambiguous.
join_on = and_(
before.c[entity.key] == after.c[entity.key],
before.c[attribute.key] == after.c[attribute.key],
)
joined = before.join(after, join_on, full=True)
return (
select(
func.coalesce(after.c[entity.key], before.c[entity.key]).label(
entity.key
),
func.coalesce(
after.c[attribute.key], before.c[attribute.key]
).label(attribute.key),
before.c.value.label("value_before"),
after.c.value.label("value_after"),
)
.select_from(joined)
.where(before.c.value.is_distinct_from(after.c.value))
.order_by(
func.coalesce(after.c[entity.key], before.c[entity.key]),
func.coalesce(after.c[attribute.key], before.c[attribute.key]),
)
)