"""EAV dimension resource factory."""
from __future__ import annotations
from dataclasses import dataclass
from pathlib import Path
from typing import TYPE_CHECKING, ClassVar
from sqlalchemy import (
BigInteger,
CheckConstraint,
Column,
DateTime,
ForeignKey,
FromClause,
Integer,
Label,
Select,
Table,
Text,
func,
literal,
select,
)
from sqlalchemy import cast as sa_cast
from sqlalchemy import types as sa_types
from codegen_database.errors import CodegenDatabaseValidationError
from codegen_database.ext.audit import ActivityViewPlugin
from codegen_database.factory.base import ResourceFactory
from codegen_database.plugin import (
Dynamic,
Plugin,
PluginOrCollection,
produces,
requires,
singleton,
)
from codegen_database.plugins.check import _CheckPlugin
from codegen_database.plugins.column_name import construct_column_name_plugin
from codegen_database.plugins.protect import RawTableProtectionPlugin
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
from codegen_database.utils.trigger import register_view_triggers
if TYPE_CHECKING:
from collections.abc import Callable
from codegen_database.factory.context import FactoryContext
_TEMPLATES = (
Path(__file__).resolve().parents[2] / "plugins" / "templates" / "eav"
)
_CHECK_TEMPLATES = (
Path(__file__).resolve().parents[2] / "plugins" / "templates" / "check"
)
_NAMING_DEFAULTS = {
"eav_entity": "%(table_name)s_entity",
"eav_attribute": "%(table_name)s_attribute",
"eav_function": "%(schema)s_%(table_name)s_%(op)s",
"eav_trigger": "%(schema)s_%(table_name)s_%(op)s",
}
_CHECK_NAMING_DEFAULTS = {
"check_function": ("_check_%(schema)s_%(table_name)s_%(op)s"),
"check_trigger": ("_check_%(schema)s_%(table_name)s_%(op)s"),
}
@dataclass
class _EAVMapping:
"""Internal mapping from a dimension name to EAV storage."""
attribute_name: str
value_column: str
column_type: sa_types.TypeEngine
nullable: bool = True
def _resolve_value_column(
col: Column,
) -> tuple[str, sa_types.TypeEngine]:
col_type = type(col.type)
col_name = f"{col_type.__name__.lower()}_value"
return col_name, col.type
def _build_eav_mappings(
dimensions: list,
) -> list[_EAVMapping]:
mappings: list[_EAVMapping] = []
for dim in dimensions:
if isinstance(dim, Column):
value_col, col_type = _resolve_value_column(dim)
mappings.append(
_EAVMapping(
attribute_name=dim.key,
value_column=value_col,
column_type=col_type,
# dim.nullable is Optional[bool]; None means
# the Column was declared without an explicit
# nullable argument, which SQLAlchemy treats
# as nullable=True.
nullable=dim.nullable is not False,
)
)
return mappings
def _needed_value_columns(
mappings: list[_EAVMapping],
) -> dict[str, sa_types.TypeEngine]:
cols: dict[str, sa_types.TypeEngine] = {}
for mapping in mappings:
if mapping.value_column not in cols:
cols[mapping.value_column] = mapping.column_type
return cols
def _pivot_aggregate(
subquery: FromClause,
mapping: _EAVMapping,
) -> Label:
col = subquery.c[mapping.value_column]
condition = subquery.c.attribute_name == literal(mapping.attribute_name)
# Postgres has no max() for uuid (or boolean), so we cast through
# a type max() does support and cast back. ``rn = 1`` already
# picks a single row per (entity, attribute) so the aggregate is
# really just an unwrap; max-via-cast is a portable way to do that.
if isinstance(mapping.column_type, sa_types.Boolean):
return sa_cast(
func.max(sa_cast(col, Integer)).filter(condition),
sa_types.Boolean(),
).label(mapping.attribute_name)
if isinstance(mapping.column_type, sa_types.Uuid):
return sa_cast(
func.max(sa_cast(col, sa_types.Text)).filter(condition),
sa_types.Uuid(),
).label(mapping.attribute_name)
return func.max(col).filter(condition).label(mapping.attribute_name)
def _construct_pivot_query( # noqa: PLR0913
entity_table: Table,
attribute_table: Table,
mappings: list[_EAVMapping],
created_at_col: str = "created_at",
updated_at_col: str = "updated_at",
deleted_at_col: str | None = None,
) -> Select:
attr = attribute_table
row_num = (
func.row_number()
.over(
partition_by=[
attr.c.entity_id,
attr.c.attribute_name,
],
order_by=[
attr.c.created_at.desc(),
attr.c.id.desc(),
],
)
.label("rn")
)
latest = (
select(attr, row_num)
.where(attr.c.attribute_name.in_([m.attribute_name for m in mappings]))
.cte("latest")
)
latest_current = select(latest).where(latest.c.rn == 1).subquery("cur")
pivot_cols = [_pivot_aggregate(latest_current, m) for m in mappings]
query = select(
entity_table.c.id.label("id"),
entity_table.c[created_at_col].label(created_at_col),
func.max(latest_current.c.created_at).label(updated_at_col),
*pivot_cols,
).select_from(
entity_table.join(
latest_current,
latest_current.c.entity_id == entity_table.c.id,
isouter=True,
)
)
# Soft delete: hide entities flagged on the entity table. This must
# be a query-level WHERE (not a post-hoc append onto the view text)
# because the pivot ends in GROUP BY.
if deleted_at_col is not None:
query = query.where(entity_table.c[deleted_at_col].is_(None))
return query.group_by(
entity_table.c.id,
entity_table.c[created_at_col],
)
def _make_eav_query_builder(
entity_key: str,
attribute_key: str,
mappings_key: str,
) -> Callable[[FactoryContext], str]:
"""Return a query builder for an EAV pivot view."""
def build(ctx: FactoryContext) -> str:
mappings: list[_EAVMapping] = ctx[mappings_key]
entity_table = ctx[entity_key]
attribute_table = ctx[attribute_key]
created_at_col = ctx["created_at_column"]
updated_at_col = ctx["updated_at_column"]
return compile_query(
_construct_pivot_query(
entity_table,
attribute_table,
mappings,
created_at_col,
updated_at_col,
ctx.get("deleted_at_column"),
)
)
return build
def _make_eav_proxy_builder(
mappings_key: str,
) -> Callable[[FactoryContext], list[Column]]:
"""Return a proxy builder for an EAV pivot view."""
def build(ctx: FactoryContext) -> list[Column]:
mappings: list[_EAVMapping] = ctx[mappings_key]
pk_col_name = ctx.pk_column_name
created_at_col = ctx["created_at_column"]
updated_at_col = ctx["updated_at_column"]
return [
ctx["pk_columns"].make_derived(name=pk_col_name),
Column(created_at_col, DateTime(timezone=True)),
Column(updated_at_col, DateTime(timezone=True)),
*[
Column(mapping.attribute_name, mapping.column_type)
for mapping in mappings
],
]
return build
def _make_eav_ops_builder(
entity_key: str,
attribute_key: str,
mappings_key: str,
) -> Callable[[FactoryContext], list[TriggerOp]]:
"""Return an ops builder for EAV dimensions."""
def build(ctx: FactoryContext) -> list[TriggerOp]:
mappings: list[_EAVMapping] = ctx[mappings_key]
entity_table = ctx[entity_key]
attribute_table = ctx[attribute_key]
entity_fullname = f"{ctx.schemaname}.{entity_table.name}"
attr_fullname = f"{ctx.schemaname}.{attribute_table.name}"
template_vars = {
"entity_table": entity_fullname,
"attr_table": attr_fullname,
"mappings": [
(
mapping.attribute_name,
mapping.value_column,
mapping.nullable,
)
for mapping in mappings
],
}
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 build
[docs]
@produces(
Dynamic("entity_key"),
Dynamic("attribute_key"),
Dynamic("mappings_key"),
)
@requires(
"pk_columns",
"created_at_column",
"deleted_at_column",
)
@singleton("__table__")
class EAVTablePlugin(Plugin):
"""Create entity and attribute tables for an EAV dimension.
Args:
entity_key: Key in ``ctx`` for the entity root table
(default ``"entity"``).
attribute_key: Key in ``ctx`` for the attribute log
(default ``"attribute"``).
mappings_key: Key in ``ctx`` for the EAV mappings list,
shared with the view and trigger plugins
(default ``"eav_mappings"``).
"""
def __init__(
self,
entity_key: str = "entity",
attribute_key: str = "attribute",
mappings_key: str = "eav_mappings",
) -> None:
"""Store the context keys."""
self.entity_key = entity_key
self.attribute_key = attribute_key
self.mappings_key = mappings_key
[docs]
def run(self, ctx: FactoryContext) -> None:
"""Create entity and attribute tables."""
pk_col_name = ctx.pk_column_name
pk_columns = ctx["pk_columns"]
created_at_col = ctx["created_at_column"]
# The soft-delete flag lives on the entity table, not as an EAV
# attribute, so drop it from the columns that become attributes.
deleted_at_col = ctx.get("deleted_at_column")
mapping_cols = [c for c in ctx.columns if c.key != deleted_at_col]
mappings = _build_eav_mappings(mapping_cols)
ctx[self.mappings_key] = mappings
entity_soft_delete_cols = (
[Column(deleted_at_col, DateTime(timezone=True), nullable=True)]
if deleted_at_col is not None
else []
)
entity_name = resolve_name(
ctx.metadata,
"eav_entity",
{
"table_name": ctx.tablename,
"schema": ctx.schemaname,
},
_NAMING_DEFAULTS,
)
entity_table = Table(
entity_name,
ctx.metadata,
pk_columns.make_derived(name=pk_col_name),
Column(
created_at_col,
DateTime(timezone=True),
server_default=func.now(),
),
*entity_soft_delete_cols,
schema=ctx.schemaname,
)
ctx[self.entity_key] = entity_table
if deleted_at_col is not None:
# The pivot view filters soft-deleted entities in-query (see
# _construct_pivot_query); tell SoftDeletePluginAfterRoot not
# to also append a WHERE onto the GROUP BY'd view text.
ctx["soft_delete_view_filtered"] = True
attr_name = resolve_name(
ctx.metadata,
"eav_attribute",
{
"table_name": ctx.tablename,
"schema": ctx.schemaname,
},
_NAMING_DEFAULTS,
)
value_cols = _needed_value_columns(mappings)
value_col_items = [
Column(name, col_type, nullable=True)
for name, col_type in value_cols.items()
]
cols_list = ", ".join(value_cols)
check = CheckConstraint(
f"num_nonnulls({cols_list}) = 1",
name=f"{attr_name}_one_value_ck",
)
entity_fq = f"{ctx.schemaname}.{entity_name}.{pk_col_name}"
# The attribute log is internal storage; users only see it via
# the pivot view. Use a BigInteger surrogate to avoid UUID's
# storage and B-tree fragmentation costs on a high-insert log
# table. The user's PK plugin still governs the public-facing
# entity and view.
attribute_table = Table(
attr_name,
ctx.metadata,
Column(pk_col_name, BigInteger, primary_key=True),
Column(
"entity_id",
ForeignKey(entity_fq, ondelete="CASCADE"),
nullable=False,
),
Column("attribute_name", Text, nullable=False),
*value_col_items,
Column(
"created_at",
DateTime(timezone=True),
server_default=func.now(),
),
check,
schema=ctx.schemaname,
)
ctx[self.attribute_key] = attribute_table
[docs]
def EAVViewPlugin( # noqa: N802
entity_key: str = "entity",
attribute_key: str = "attribute",
mappings_key: str = "eav_mappings",
primary_key: str = "primary",
) -> ViewPlugin:
"""Create a configured ViewPlugin for EAV dimensions.
Args:
entity_key: Key in ``ctx`` for the entity root table
(default ``"entity"``).
attribute_key: Key in ``ctx`` for the attribute log
(default ``"attribute"``).
mappings_key: Key in ``ctx`` for the EAV mappings list
(default ``"eav_mappings"``).
primary_key: Key in ``ctx`` to store the view proxy
under (default ``"primary"``).
Returns:
A :class:`~codegen_database.plugins.view.ViewPlugin` configured
for EAV pivot views.
"""
return ViewPlugin(
query_builder=_make_eav_query_builder(
entity_key, attribute_key, mappings_key
),
proxy_builder=_make_eav_proxy_builder(mappings_key),
primary_key=primary_key,
extra_requires=[
entity_key,
attribute_key,
mappings_key,
"pk_columns",
"created_at_column",
"updated_at_column",
],
)
def _validate_trigger_ordering(
ctx: FactoryContext,
view_fullname: str,
) -> None:
"""Validate check triggers fire before existing triggers.
Postgres fires multiple INSTEAD OF triggers on the same view
in alphabetical order by trigger name. Check triggers must
sort before the main dimension triggers so that constraint
violations are caught before any data is modified.
Args:
ctx: Factory context (for metadata access).
view_fullname: Fully qualified view name to check.
Raises:
CodegenDatabaseValidationError: If a check trigger name sorts
after an existing non-check trigger on the same
view.
"""
triggers_info = ctx.metadata.info.get("triggers")
if triggers_info is None:
return
check_triggers: list[str] = []
other_triggers: list[str] = []
for t in triggers_info.triggers:
if t.on != view_fullname:
continue
name = t.name
if name.startswith("_check_"):
check_triggers.append(name)
else:
other_triggers.append(name)
for ct in check_triggers:
for ot in other_triggers:
if ct > ot:
msg = (
f"Check trigger {ct!r} sorts after "
f"existing trigger {ot!r} on "
f"{view_fullname}. Postgres fires "
f"multiple INSTEAD OF triggers in "
f"alphabetical order by name, so "
f"check triggers must sort before "
f"the main dimension triggers to "
f"enforce constraints before data "
f"is modified. Consider renaming "
f"the check or using a naming "
f"convention that ensures the "
f"check trigger sorts first "
f"(e.g. a '_check_' prefix)."
)
raise CodegenDatabaseValidationError(msg)
[docs]
class TriggerCheckPlugin(_CheckPlugin):
"""Enforce checks via INSTEAD OF triggers (EAV dimensions).
Generates a single trigger function per view per operation
(INSERT/UPDATE) that validates all
:class:`~codegen_database.check.CodegenDatabaseCheck`
items. Triggers use a ``_check_`` prefix to fire before the
main dimension triggers (Postgres fires multiple INSTEAD OF
triggers in alphabetical order by name).
Args:
table_key: Key in ``ctx`` for the trigger target view
(default ``"primary"``).
"""
def _column_names(self, ctx: FactoryContext) -> set[str]:
"""Return column names from the virtual (schema_items) columns."""
return {col.key for col in ctx.columns}
def _apply(
self,
ctx: FactoryContext,
checks: list,
) -> None:
"""Register INSTEAD OF trigger functions for each check."""
resolved_checks = [
(cave_check.resolve(lambda c: f"NEW.{c}"), cave_check.name)
for cave_check in checks
]
template = load_template(_CHECK_TEMPLATES / "validate.plpgsql.mako")
body = template.render(checks=resolved_checks)
if self.table_key not in ctx:
return
view = ctx[self.table_key]
view_schema = view.schema or ctx.schemaname
view_fullname = f"{view_schema}.{ctx.tablename}"
register_view_triggers(
metadata=ctx.metadata,
view_schema=view_schema,
view_fullname=view_fullname,
tablename=ctx.tablename,
ops=[
("insert", body),
("update", body),
],
naming_defaults=_CHECK_NAMING_DEFAULTS,
function_key="check_function",
trigger_key="check_trigger",
)
_validate_trigger_ordering(ctx, view_fullname)
[docs]
class CodegenDatabaseEAV(ResourceFactory):
"""Create an EAV (Entity-Attribute-Value) dimension.
Internal plugins (always present), in order:
The column-name plugin
:func:`~codegen_database.plugins.column_name.construct_column_name_plugin`
sets the ``created_at`` column name. Then:
1. :class:`EAVTablePlugin` -- entity + attribute tables.
2. :class:`EAVViewPlugin` -- pivot view proxy.
3. :class:`TriggerCheckPlugin` -- trigger-based checks on
the pivot view.
4. :class:`~codegen_database.plugins.trigger.InsteadOfTriggerPlugin`
-- INSTEAD OF triggers (activates when a view plugin
produces ``"primary"``).
A :class:`~codegen_database.plugins.pk.SerialPKPlugin` is auto-added
when no user plugin produces ``pk_columns``.
"""
_FK_TARGET_KEY: ClassVar[str] = "entity"
_INTERNAL_PLUGINS: ClassVar[list[PluginOrCollection]] = [
construct_column_name_plugin("created_at_column", "created_at"),
construct_column_name_plugin("updated_at_column", "updated_at"),
EAVTablePlugin(),
EAVViewPlugin(),
TriggerCheckPlugin(),
RawTableProtectionPlugin("entity", "attribute"),
ActivityViewPlugin(),
InsteadOfTriggerPlugin(
ops_builder=_make_eav_ops_builder(
"entity", "attribute", "eav_mappings"
),
naming_defaults=_NAMING_DEFAULTS,
function_key="eav_function",
trigger_key="eav_trigger",
view_key="primary",
extra_requires=[
"entity",
"attribute",
"eav_mappings",
],
),
]