Source code for codegen_database.config

"""Global codegen_database configuration."""

from __future__ import annotations

from dataclasses import dataclass, field
from typing import TYPE_CHECKING

from codegen_database.extension import (
    discover_extensions,
    validate_extension_deps,
)

if TYPE_CHECKING:
    from sqlalchemy import MetaData

    from codegen_database.extension import CodegenDatabaseExtension
    from codegen_database.plugin import Plugin

#: Default schema for codegen_database-managed utility objects -- the
#: ``CodegenDatabaseConfig.utility_schema`` default and the fallback
#: :func:`resolve_utility_schema` uses when no config is present.
DEFAULT_UTILITY_SCHEMA = "codegen_database"




@dataclass
class CodegenDatabaseConfig:
    """Global plugin and extension registry.

    Plugins registered here are prepended to every factory's resolved
    plugin list, so they run before factory-specific plugins.

    Extensions bundle plugins, metadata hooks, Alembic hooks, and CLI
    commands into a single unit.

    Example::

        config = CodegenDatabaseConfig()
        config.register(TimestampPlugin(), TenantPlugin())

        CodegenDatabaseSimple(
            "users", "public", metadata, ..., config=config
        )

    Args:
        plugins: Global plugins prepended to every factory.
        extensions: Manually registered extension instances.
        auto_discover: Whether to discover extensions via
            entry points.  Defaults to ``True``.
        utility_schema: PostgreSQL schema for codegen_database-managed
            utility functions (e.g. ``ledger_apply_state``).
            Defaults to ``"codegen_database"``.  Override only if your
            project already uses a schema named ``"codegen_database"``.

    """

    plugins: list[Plugin] = field(default_factory=list)
    extensions: list[CodegenDatabaseExtension] = field(default_factory=list)
    auto_discover: bool = True
    utility_schema: str = DEFAULT_UTILITY_SCHEMA



    def register(self, *plugins: Plugin) -> CodegenDatabaseConfig:
        """Register one or more plugins globally.

        Args:
            *plugins: Plugin instances to add.

        Returns:
            ``self`` for chaining.

        """
        self.plugins.extend(plugins)
        return self




    def use(
        self, *extensions: CodegenDatabaseExtension
    ) -> CodegenDatabaseConfig:
        """Register one or more extensions.

        Args:
            *extensions: Extension instances to add.

        Returns:
            ``self`` for chaining.

        """
        self.extensions.extend(extensions)
        return self


    def _resolved_extensions(self) -> list[CodegenDatabaseExtension]:
        """Return manual + discovered extensions, deduped by name.

        Manual extensions take precedence over discovered ones
        with the same name.

        Returns:
            Ordered list of extension instances.

        """
        seen: dict[str, CodegenDatabaseExtension] = {}

        for ext in self.extensions:
            seen[ext.name] = ext

        if self.auto_discover:
            for name, ext_cls in discover_extensions().items():
                if name not in seen:
                    seen[name] = ext_cls(name=name)

        result = list(seen.values())
        validate_extension_deps(result)
        return result

    @property
    def all_plugins(self) -> list[Plugin]:
        """Return extension plugins + direct plugins.

        Extension plugins are prepended before direct plugins.

        Returns:
            Combined plugin list.

        """
        ext_plugins: list[Plugin] = []

        for ext in self._resolved_extensions():
            ext_plugins.extend(ext.plugins())

        return ext_plugins + self.plugins





def resolve_utility_schema(metadata: MetaData) -> str:
    """Return the codegen_database utility schema for *metadata*.

    Reads ``utility_schema`` off the :class:`CodegenDatabaseConfig`
    registered at ``metadata.info["codegen_database_config"]``, falling back
    to :data:`DEFAULT_UTILITY_SCHEMA` when no config is present.

    Any codegen_database-managed object that needs a shared home -- utility
    functions, the autogenerated-identifier pointer table -- routes
    through here so a single ``utility_schema`` setting governs them
    all.

    Args:
        metadata: SQLAlchemy ``MetaData`` to inspect.

    Returns:
        The schema name for codegen_database-managed utility objects.

    """
    config = metadata.info.get("codegen_database_config")

    if config is None:
        return DEFAULT_UTILITY_SCHEMA

    return getattr(config, "utility_schema", DEFAULT_UTILITY_SCHEMA)