Source code for codegen_database.validator

"""Schema item validators for dimension factories."""

from __future__ import annotations

import re
from typing import TYPE_CHECKING, Protocol

from sqlalchemy.schema import DefaultClause

from codegen_database.errors import CodegenDatabaseValidationError

if TYPE_CHECKING:
    from collections.abc import Iterable

    from sqlalchemy import Table
    from sqlalchemy.schema import SchemaItem

    from codegen_database.check import CodegenDatabaseCheck
    from codegen_database.index import CodegenDatabaseIndex

    _SchemaItemType = SchemaItem | CodegenDatabaseCheck | CodegenDatabaseIndex


_FUNCTION_CALL_DEFAULT = re.compile(
    r'^\s*[A-Za-z_][\w.]*\(.*\)(::[\w .\[\]"]+)?\s*$', re.DOTALL
)




def frozen_function_default(server_default: object) -> str | None:
    """Return the SQL of a ``server_default`` that PostgreSQL will freeze.

    A ``server_default`` given as a plain :class:`str` is emitted as a
    *quoted* SQL literal (``DEFAULT 'now()'``), which PostgreSQL evaluates
    once at DDL time -- every row then shares that single frozen value.
    When the string is shaped like a function call (``now()``,
    ``gen_random_uuid()``, ``nextval(...)`` ...) that is almost never what
    was meant.  Wrap it in ``func.<fn>()`` or ``text(...)`` so it renders
    as a live call re-evaluated per row.

    Returns the offending string, or ``None`` when the default is safe (a
    SQL element / ``text()`` clause, or a plain literal constant such as
    ``"0"`` / ``"true"``).
    """
    arg = (
        server_default.arg
        if isinstance(server_default, DefaultClause)
        else server_default
    )

    if isinstance(arg, str) and _FUNCTION_CALL_DEFAULT.match(arg):
        return arg

    return None





def reject_frozen_function_defaults(tables: Iterable[Table]) -> None:
    """Raise if any column defaults a function call passed as a plain string.

    Guards against the silent ``DEFAULT 'now()'`` bug (frozen at migration
    time) across every column in *tables* -- both user-supplied columns and
    the timestamp columns the built-in plugins inject.
    """
    for table in tables:
        for col in table.columns:
            frozen = frozen_function_default(col.server_default)

            if frozen is not None:
                msg = (
                    f"Column {col.name!r} on table {table.fullname!r} has "
                    f"server_default={frozen!r} as a plain string, which "
                    f"renders as the SQL literal DEFAULT '{frozen}'.  "
                    f"PostgreSQL evaluates that once at migration time, so "
                    f"every row gets the same frozen value.  Pass a SQL "
                    f"expression instead, e.g. func.{frozen} or "
                    f'text("{frozen}").'
                )
                raise CodegenDatabaseValidationError(msg)



class _SchemaItemValidator(Protocol):
    def __call__(self, item: _SchemaItemType) -> bool: ...




def is_schema_item_not_primary_key(
    item: _SchemaItemType,
) -> bool:
    """Return True if the item is not a primary key column.

    Args:
        item: The schema item to inspect.

    Returns:
        ``True`` if *item* is not a primary key column.

    """
    return not getattr(item, "primary_key", False)



_default_validators: list[_SchemaItemValidator] = [
    is_schema_item_not_primary_key,
]




def validate_schema_items(
    items: list,
    *,
    validators: list[_SchemaItemValidator] | None = None,
) -> None:
    """Validate a list of SchemaItems against the given validators.

    Args:
        items: Schema items to validate.
        validators: Validators to run; defaults to
            ``[is_schema_item_not_primary_key]``.

    Raises:
        CodegenDatabaseValidationError: If any item fails a validator.

    """
    validators = validators or _default_validators

    for item in items:
        for validator in validators:
            if not validator(item):
                name = getattr(validator, "__name__", repr(validator))
                msg = f"{name} failed for {item!r}"
                raise CodegenDatabaseValidationError(msg)