"""Core ResourceFactory: plugin runner."""
from __future__ import annotations
from graphlib import CycleError, TopologicalSorter
from typing import TYPE_CHECKING, ClassVar
from codegen_database.errors import CodegenDatabaseValidationError
from codegen_database.factory.context import FactoryContext
from codegen_database.fk import DimensionRef, register_dimension
from codegen_database.plugin import Plugin, PluginOrCollection
from codegen_database.plugins.autogenerated_identifier import (
AutoIdentifierPlugin,
)
from codegen_database.plugins.check import TableCheckPlugin, _CheckPlugin
from codegen_database.plugins.fk import TableFKPlugin
from codegen_database.plugins.index import TableIndexPlugin
from codegen_database.plugins.pk import SerialPKPlugin
from codegen_database.plugins.timestamps import TimestampPlugin
from codegen_database.validator import (
reject_frozen_function_defaults,
validate_schema_items,
)
if TYPE_CHECKING:
from collections.abc import Callable, Generator
from sqlalchemy import MetaData
from sqlalchemy.schema import SchemaItem
from sqlalchemy.sql.expression import FromClause
from codegen_database.check import CodegenDatabaseCheck
from codegen_database.index import CodegenDatabaseIndex
def _has_pk_producer(plugins: list[Plugin]) -> bool:
"""Return True if any plugin produces ``pk_columns``."""
return any("pk_columns" in p.resolved_produces() for p in plugins)
def _has_timestamp_producer(plugins: list[Plugin]) -> bool:
"""Return True if any plugin produces a timestamp ctx key.
Either ``created_at_column`` or ``updated_at_column`` counts
-- factories that want one but not the other (e.g. legacy
append-only resources that only track ``created_at``) can
register that single key without triggering the auto-add.
"""
keys = {"created_at_column", "updated_at_column"}
return any(keys & set(p.resolved_produces()) for p in plugins)
def _has_instance(plugins: list[Plugin], cls: type) -> bool:
"""Return True if any plugin is an instance of *cls*."""
return any(isinstance(p, cls) for p in plugins)
def _auto_add_defaults(plugins: list[Plugin]) -> list[Plugin]:
"""Auto-add the column-scanning built-in plugins.
``TableCheckPlugin``, ``TableIndexPlugin``, ``TableFKPlugin``
and ``AutoIdentifierPlugin`` are added with default settings
when no instance of their class (or, for checks, any
``_CheckPlugin`` subclass) is already present. Each is a no-op
when the corresponding schema items are absent, so adding them
unconditionally is safe.
"""
added: list[Plugin] = []
if not _has_instance(plugins, _CheckPlugin):
added.append(TableCheckPlugin())
if not _has_instance(plugins, TableIndexPlugin):
added.append(TableIndexPlugin())
if not _has_instance(plugins, TableFKPlugin):
added.append(TableFKPlugin())
if not _has_instance(plugins, AutoIdentifierPlugin):
added.append(AutoIdentifierPlugin())
return plugins + added
def _deconstruct_plugin_collections(
plugins: list[PluginOrCollection],
) -> Generator[Plugin]:
"""Flatten plugin collections into a single list of Plugin instances."""
for plugin in plugins:
if isinstance(plugin, Plugin):
yield plugin
else:
yield from plugin
def _resolve_plugins(
config: object | None, # CodegenDatabaseConfig, avoiding circular import
plugins: list[PluginOrCollection] | None,
extra_plugins: list[PluginOrCollection] | None,
defaults: list[PluginOrCollection],
internal: list[PluginOrCollection] | None = None,
) -> list[Plugin]:
"""Resolve the effective plugin list for a factory invocation.
Args:
config: Optional
:class:`~codegen_database.config.CodegenDatabaseConfig`
providing global plugins.
plugins: If given, replaces ``defaults``. If ``None``,
``defaults`` is used.
extra_plugins: Always appended to the resolved list.
defaults: The factory's ``DEFAULT_PLUGINS``.
internal: Internal plugins always appended last.
Returns:
Ordered list of plugins to run.
"""
global_plugins = _deconstruct_plugin_collections(
getattr(
config,
"all_plugins",
getattr(config, "plugins", []),
)
)
factory_plugins = _deconstruct_plugin_collections(plugins or defaults)
local_plugins = _deconstruct_plugin_collections(extra_plugins or [])
user_plugins = [*global_plugins, *factory_plugins, *local_plugins]
# Auto-add SerialPKPlugin if no user plugin produces pk_columns
# and internal plugins need it.
internal_plugins = list(_deconstruct_plugin_collections(internal or []))
if internal_plugins and not _has_pk_producer(user_plugins):
user_plugins = [SerialPKPlugin(), *user_plugins]
# Auto-add a default-shaped TimestampPlugin if no plugin --
# user-supplied or internal -- already registers the ctx
# keys. Append-only factories ship with their own
# internal column-name plugins for the timestamp columns;
# those satisfy the check and the auto-add stays out of
# the way.
if internal_plugins and not _has_timestamp_producer(
user_plugins + internal_plugins
):
user_plugins = [TimestampPlugin(), *user_plugins]
all_plugins = user_plugins + internal_plugins
# Only auto-add check/index/fk plugins when internal plugins
# are present (i.e. there is a table-creating plugin).
if internal_plugins:
all_plugins = _auto_add_defaults(all_plugins)
return all_plugins
def _run_plugin_validators(
plugins: list[Plugin],
) -> None:
"""Collect and run all class-level validators, deduped by id.
Args:
plugins: Resolved plugin list to validate.
"""
seen: set[int] = set()
for p in plugins:
validators: list[Callable[[list[Plugin]], None]] = getattr(
type(p), "_validators", []
)
for v in validators:
if id(v) not in seen:
seen.add(id(v))
v(plugins)
def _sort_plugins(plugins: list[Plugin]) -> list[Plugin]:
"""Sort plugins topologically by produces/requires declarations.
Plugins with no declared dependencies keep their original relative
order. References to keys not produced by any plugin in this list
are treated as externally satisfied and ignored for ordering.
Args:
plugins: The resolved plugin list to sort.
Returns:
A new list of the same plugins in a valid execution order.
Raises:
CodegenDatabaseValidationError: If a dependency cycle is detected.
"""
# Last producer of a key wins for dependency resolution — if
# multiple plugins produce the same key, requires-edges point
# to the last one, meaning overriding plugins run after the
# ones they override.
producers: dict[str, Plugin] = {}
for p in plugins:
for key in p.resolved_produces():
producers[key] = p
# Build predecessor graph: each plugin maps to the set of
# plugins whose output it requires. Only edges within this
# plugin list matter.
graph: dict[Plugin, set[Plugin]] = {p: set() for p in plugins}
for p in plugins:
for key in p.resolved_requires():
producer = producers.get(key)
if producer is not None:
graph[p].add(producer)
# Use original list index as tiebreaker so unrelated plugins
# preserve their declared order.
original_order = {id(p): i for i, p in enumerate(plugins)}
ts = TopologicalSorter(graph)
try:
ts.prepare()
except CycleError as exc:
names = ", ".join(type(p).__name__ for p in exc.args[1])
msg = f"Circular plugin dependency detected among: {names}"
raise CodegenDatabaseValidationError(msg) from exc
result: list[Plugin] = []
while ts.is_active():
ready = sorted(
ts.get_ready(),
key=lambda p: original_order[id(p)],
)
for p in ready:
result.append(p)
ts.done(p)
return result
[docs]
class ResourceFactory:
"""Core factory: resolves plugins and runs them in dependency order.
Subclasses declare ``DEFAULT_PLUGINS`` for user-facing defaults
and ``_INTERNAL_PLUGINS`` for always-present built-in logic.
Callers can override or extend the plugin list via
``plugins`` / ``extra_plugins``, and inject global plugins via
``config``.
Plugin execution order is determined by each plugin's
:attr:`~codegen_database.plugin.Plugin.produces` and
:attr:`~codegen_database.plugin.Plugin.requires` declarations. Plugins
with no declared dependencies run in the order they appear in
the list.
Resolution order: ``global_plugins + user_plugins +
internal_plugins``, then topological sort.
If no user or global plugin produces ``pk_columns`` and
internal plugins are present, a
:class:`~codegen_database.plugins.pk.SerialPKPlugin` is auto-prepended.
Args:
tablename: Name of the dimension table.
schemaname: PostgreSQL schema for all generated objects.
metadata: SQLAlchemy ``MetaData`` the objects are bound to.
schema_items: Column and constraint definitions. Must not
include a primary key column.
config: Optional global config supplying prepended plugins.
plugins: If given, replaces ``DEFAULT_PLUGINS`` entirely.
extra_plugins: Appended to the resolved plugin list.
Raises:
CodegenDatabaseValidationError: If any schema item fails validation,
two plugins share a singleton group, two plugins produce
the same ctx key, or a plugin dependency cycle is
detected.
"""
DEFAULT_PLUGINS: ClassVar[list[PluginOrCollection]] = []
_INTERNAL_PLUGINS: ClassVar[list[PluginOrCollection]] = []
_FK_TARGET_KEY: ClassVar[str] = "primary"
table: FromClause
"""The root selectable created by the factory.
This is the ``__root__`` context value set by the table
plugin, exposing the column metadata for use in queries,
foreign key references, and ledger event lambdas.
"""
ctx: FactoryContext
"""The factory context after plugin execution.
Downstream view factories and query builders read this to access
tables, columns, and other plugin outputs.
"""
def __init__( # noqa: PLR0913
self,
tablename: str,
schemaname: str,
metadata: MetaData,
schema_items: list[
SchemaItem | CodegenDatabaseCheck | CodegenDatabaseIndex
],
*,
config: object | None = None,
plugins: list[PluginOrCollection] | None = None,
extra_plugins: list[PluginOrCollection] | None = None,
) -> None:
"""Create the dimension and register it on *metadata*."""
validate_schema_items(schema_items)
if (
config is not None
and "codegen_database_config" not in metadata.info
):
metadata.info["codegen_database_config"] = config
resolved = _resolve_plugins(
config,
plugins,
extra_plugins,
self.DEFAULT_PLUGINS,
self._INTERNAL_PLUGINS,
)
_run_plugin_validators(resolved)
ctx = FactoryContext(
tablename=tablename,
schemaname=schemaname,
metadata=metadata,
schema_items=list(schema_items),
plugins=resolved,
)
tables_before = set(metadata.tables)
for plugin in _sort_plugins(resolved):
plugin.run(ctx)
reject_frozen_function_defaults(
table
for name, table in metadata.tables.items()
if name not in tables_before
)
self.ctx = ctx
if "__root__" in ctx:
self.table = ctx["__root__"]
if self._FK_TARGET_KEY in ctx:
fk_table = ctx[self._FK_TARGET_KEY]
register_dimension(
metadata,
tablename,
DimensionRef(
schema=schemaname,
table=fk_table.name,
),
)