"""Generic INSTEAD OF trigger plugin."""
from __future__ import annotations
from dataclasses import dataclass, field
from typing import TYPE_CHECKING, Any
from codegen_database.plugin import Dynamic, Plugin, requires
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 pathlib import Path
from codegen_database.factory.context import FactoryContext
[docs]
@dataclass
class TriggerOp:
"""A single INSTEAD OF trigger operation.
Args:
name: DML operation name (``"insert"``, ``"update"``,
``"delete"``).
body: Pre-rendered PL/pgSQL function body.
"""
name: str
body: str
# Context key under which a plugin may inject a delete-trigger override.
DELETE_TRIGGER_OVERRIDE_KEY = "delete_trigger_override"
[docs]
@dataclass(frozen=True)
class DeleteTriggerOverride:
"""Replacement template for a dimension's INSTEAD OF DELETE body.
A plugin (e.g.
:class:`~codegen_database.plugins.soft_delete.SoftDeletePlugin`)
injects this into the factory context under
``DELETE_TRIGGER_OVERRIDE_KEY`` to swap the default physical-delete
template for an alternative (e.g. a soft-delete ``UPDATE``). Each
factory renders its delete op through :func:`render_delete_op`, so no
factory needs to special-case the override itself.
Args:
template_name: Filename of the delete template to render
instead of ``delete.plpgsql.mako``, resolved within the
calling factory's own template directory.
template_vars: Extra template variables merged on top of the
factory's base delete vars (e.g. ``deleted_at_column``).
"""
template_name: str
template_vars: dict[str, Any] = field(default_factory=dict)
[docs]
def render_delete_op(
ctx: FactoryContext,
templates_dir: Path,
base_vars: dict[str, Any],
) -> TriggerOp:
"""Render the DELETE :class:`TriggerOp`, honoring a context override.
Renders ``delete.plpgsql.mako`` from *templates_dir* with *base_vars*
unless a plugin injected a :class:`DeleteTriggerOverride` in ``ctx``,
in which case the override's template and extra vars take precedence.
This keeps soft-delete (and any future delete variant) out of each
factory's ops builder.
Args:
ctx: The factory context.
templates_dir: The calling factory's template directory.
base_vars: The factory's default delete template variables.
Returns:
The rendered ``"delete"`` :class:`TriggerOp`.
"""
override: DeleteTriggerOverride | None = ctx.get(
DELETE_TRIGGER_OVERRIDE_KEY
)
template_name = (
override.template_name if override else "delete.plpgsql.mako"
)
template_vars = dict(base_vars)
if override is not None:
template_vars.update(override.template_vars)
return TriggerOp(
"delete",
load_template(templates_dir / template_name).render(**template_vars),
)
[docs]
@requires(Dynamic("view_key"))
class InsteadOfTriggerPlugin(Plugin):
"""Register INSTEAD OF triggers from pre-rendered PL/pgSQL bodies.
A generic, composable trigger plugin. Each factory type
provides an ``ops_builder`` callable that reads from the
factory context and returns a list of :class:`TriggerOp`
with fully rendered PL/pgSQL.
Args:
ops_builder: Callable that takes a
:class:`~codegen_database.factory.context.FactoryContext` and
returns a list of :class:`TriggerOp`.
naming_defaults: Default naming templates for function
and trigger names.
function_key: Key for function name resolution.
trigger_key: Key for trigger name resolution.
view_key: Key in ``ctx`` for the trigger target view.
If absent from ``ctx``, trigger registration is
skipped.
permitted_operations: When set, only operations whose
names appear in this list are registered.
extra_requires: Additional ctx keys to declare as
dependencies for topological ordering.
"""
def __init__( # noqa: PLR0913
self,
ops_builder: Callable[[FactoryContext], list[TriggerOp]],
naming_defaults: dict[str, str],
function_key: str,
trigger_key: str,
view_key: str = "primary",
permitted_operations: list[str] | None = None,
*,
extra_requires: list[str] | None = None,
) -> None:
"""Store configuration."""
self.ops_builder = ops_builder
self.naming_defaults = naming_defaults
self.function_key = function_key
self.trigger_key = trigger_key
self.view_key = view_key
self.permitted_operations = permitted_operations
self._extra_requires = extra_requires or []
[docs]
def resolved_requires(self) -> list[str]:
"""Return base requires plus extra runtime keys."""
base = super().resolved_requires()
return base + list(self._extra_requires)
def _collect_views(
self,
ctx: FactoryContext,
) -> list[tuple[str, str]]:
"""Return ``(schema, fullname)`` pairs for trigger targets.
Returns a single-element list with the view stored at
``view_key``, or an empty list if that key is absent.
"""
if self.view_key not in ctx:
return []
view = ctx[self.view_key]
view_schema = view.schema or ctx.schemaname
return [(view_schema, f"{view_schema}.{ctx.tablename}")]
[docs]
def run(self, ctx: FactoryContext) -> None:
"""Register INSTEAD OF triggers on the target view.
Skips entirely when ``view_key`` is absent from ``ctx``.
When ``permitted_operations`` was not set at construction
time, falls back to ``ctx["permitted_operations"]`` if
present.
"""
views = self._collect_views(ctx)
if not views:
return
ops = self.ops_builder(ctx)
allowed = self.permitted_operations
if allowed is None and "permitted_operations" in ctx:
allowed = ctx["permitted_operations"]
if allowed is not None:
ops = [o for o in ops if o.name in set(allowed)]
rendered: list[tuple[str, str]] = [(o.name, o.body) for o in ops]
if not rendered:
return
for view_schema, view_fullname in views:
register_view_triggers(
metadata=ctx.metadata,
view_schema=view_schema,
view_fullname=view_fullname,
tablename=ctx.tablename,
ops=rendered,
naming_defaults=self.naming_defaults,
function_key=self.function_key,
trigger_key=self.trigger_key,
)