"""Build EDRI validators from SQLAlchemy mapped columns."""
from __future__ import annotations
from collections.abc import Callable
from datetime import date, datetime, time
from inspect import signature
from typing import Any
from sqlalchemy import Enum as SQLAlchemyEnum
from sqlalchemy.orm import ColumnProperty
from sqlalchemy.sql import ColumnElement
from edri.utility.coercion import TypeCoercer
from edri.utility.validation import (
DateTimeValidator,
DateValidator,
FloatValidator,
IntegerValidator,
StringValidator,
TimeValidator,
)
__all__ = ["orm_validator"]
_TYPE_COERCER = TypeCoercer()
_EDRI_VALIDATORS: dict[type[Any], Callable[..., Any]] = {
str: StringValidator,
int: IntegerValidator,
float: FloatValidator,
date: DateValidator,
datetime: DateTimeValidator,
time: TimeValidator,
}
def _column_label(attribute: Any, column: ColumnElement[Any]) -> str:
owner = getattr(attribute, "class_", None)
key = getattr(attribute, "key", None) or getattr(column, "key", None)
if owner is not None and key:
return f"{owner.__name__}.{key}"
table = getattr(column, "table", None)
table_name = getattr(table, "name", None)
if table_name and key:
return f"{table_name}.{key}"
return str(key or column)
def _resolve_column(attribute: Any) -> ColumnElement[Any]:
if isinstance(attribute, ColumnElement):
return attribute
prop = getattr(attribute, "property", None)
if not isinstance(prop, ColumnProperty):
raise TypeError(
"orm_validator() expects a SQLAlchemy mapped column attribute or "
"table column"
)
columns = tuple(prop.columns)
if len(columns) != 1 or not isinstance(columns[0], ColumnElement):
raise TypeError(
"orm_validator() requires an attribute mapped to exactly one column"
)
return columns[0]
def _validator_keywords(validator: Callable[..., Any]) -> frozenset[str]:
"""Keyword parameter names a built-in EDRI validator accepts."""
return frozenset(
name
for name, parameter in signature(validator).parameters.items()
if name not in ("self", "value", "data")
and parameter.kind
in (parameter.POSITIONAL_OR_KEYWORD, parameter.KEYWORD_ONLY)
)
[docs]
def orm_validator(attribute: Any, /, **overrides: Any) -> Callable[[Any], Any]:
"""Return a validator callable derived from one ORM mapped column.
Pass the result to :func:`edri.utility.function.inject` in an event field::
name: inject(orm_validator(User.name))
The column's ``python_type`` selects the matching built-in EDRI validator.
A bounded SQLAlchemy string also supplies ``maximum_length``; a
string-backed ``SQLAlchemyEnum("a", "b")`` supplies its allowed values.
Types without a dedicated EDRI validator — including a column mapped to a
Python ``enum.Enum``, which is coerced against that class — use EDRI's
normal scalar coercion.
``overrides`` are forwarded to that built-in validator, taking precedence
over anything derived from the column — ``orm_validator(User.name,
maximum_length=3, minimum_length=2)`` tightens the column's own length.
An override the underlying validator does not accept, or any override on a
column with no dedicated validator (UUID, ``Decimal``, ``bool``, a Python
enum), is rejected here rather than silently ignored.
Request optionality stays explicit at the event boundary; use
``inject(orm_validator(Model.field)) | None = None`` when ``None`` is valid.
Database-backed rules such as uniqueness and foreign keys are intentionally
outside this scalar validator.
"""
column = _resolve_column(attribute)
column_type = column.type
label = _column_label(attribute, column)
# SQLAlchemy < 2.1 raises NotImplementedError for a type without a known
# Python type; 2.1 returns ``object`` instead.
try:
python_type = column_type.python_type
except (AttributeError, NotImplementedError) as error:
raise TypeError(
f"Cannot derive a Python type for SQLAlchemy column {label}"
) from error
if python_type is object:
raise TypeError(f"Cannot derive a Python type for SQLAlchemy column {label}")
if not isinstance(python_type, type):
raise TypeError(
f"SQLAlchemy column {label} exposes invalid python_type "
f"{python_type!r}"
)
edri_validator = _EDRI_VALIDATORS.get(python_type)
validator_parameters: dict[str, Any] = {}
# A SQLAlchemy Enum subclasses String and auto-computes ``.length`` from its
# longest member; that is an internal detail, not a length bound the schema
# asked for, and the allowed-values check already constrains the field.
if edri_validator is StringValidator and not isinstance(column_type, SQLAlchemyEnum):
length = getattr(column_type, "length", None)
if length is not None:
if isinstance(length, bool) or not isinstance(length, int):
raise TypeError(
f"SQLAlchemy column {label} exposes invalid string length "
f"{length!r}"
)
validator_parameters["maximum_length"] = length
if overrides:
if edri_validator is None:
raise TypeError(
f"orm_validator() for {label} derives no built-in validator "
f"(python type {python_type.__name__}); it takes no override "
f"parameters, got {sorted(overrides)!r}"
)
allowed = _validator_keywords(edri_validator)
unknown = sorted(set(overrides) - allowed)
if unknown:
raise TypeError(
f"orm_validator() for {label}: {edri_validator.__name__} does "
f"not accept {unknown!r}; accepts {sorted(allowed)!r}"
)
validator_parameters.update(overrides)
# Only a value-list SQLAlchemyEnum ("a", "b") carries allowed values that
# would otherwise be re-typed in the event; a column mapped to a Python
# Enum uses that class as its own source of truth through coercion.
allowed_values: tuple[str, ...] | None = None
if isinstance(column_type, SQLAlchemyEnum) and column_type.enum_class is None:
allowed_values = tuple(column_type.enums)
def validate(value: Any, /) -> Any:
if value is None:
raise ValueError(
f"SQLAlchemy column {label} does not accept None in this event; "
"make the injection nullable explicitly"
)
if edri_validator is None:
converted = _TYPE_COERCER.convert_type(value, python_type)
else:
converted = edri_validator(value, **validator_parameters)
if allowed_values is not None and converted not in allowed_values:
raise ValueError(
f"Invalid data provided for {label}. "
f"Expected one of {allowed_values!r}"
)
return converted
safe_label = label.replace(".", "_").replace("-", "_")
validate.__name__ = f"validate_{safe_label}"
validate.__qualname__ = validate.__name__
setattr(validate, "__edri_base_type__", python_type)
if edri_validator is not None:
setattr(validate, "__edri_validator__", edri_validator)
setattr(
validate,
"__edri_validator_parameters__",
{
**validator_parameters,
**({"allowed_values": allowed_values} if allowed_values is not None else {}),
},
)
return validate