:ticket:`12570`
+.. _change_9832:
+
+New RegistryEvents System for ORM Mapping Customization
+--------------------------------------------------------
+
+SQLAlchemy 2.1 introduces :class:`.RegistryEvents`, providing for event
+hooks that are specific to a :class:`_orm.registry`. These events include
+:meth:`_orm.RegistryEvents.before_configured` and :meth:`_orm.RegistryEvents.after_configured`
+to complement the same-named events that can be established on a
+:class:`_orm.Mapper`, as well as :meth:`_orm.RegistryEvents.resolve_type_annotation`
+that allows programmatic access to the ORM Annotated Declarative type resolution
+process. Examples are provided illustrating how to define resolution schemes
+for any kind of type hierarchy in an automated fashion, including :pep:`695`
+type aliases.
+
+E.g.::
+
+ from typing import Any
+
+ from sqlalchemy import event
+ from sqlalchemy.orm import DeclarativeBase
+ from sqlalchemy.orm import registry as RegistryType
+ from sqlalchemy.orm import TypeResolve
+ from sqlalchemy.types import TypeEngine
+
+
+ class Base(DeclarativeBase):
+ pass
+
+
+ @event.listens_for(Base, "resolve_type_annotation")
+ def resolve_custom_type(resolve_type: TypeResolve) -> TypeEngine[Any] | None:
+ if resolve_type.resolved_type is MyCustomType:
+ return MyCustomSQLType()
+ else:
+ return None
+
+
+ @event.listens_for(Base, "after_configured")
+ def after_base_configured(registry: RegistryType) -> None:
+ print(f"Registry {registry} fully configured")
+
+.. seealso::
+
+ :ref:`orm_declarative_resolve_type_event` - Complete documentation on using
+ the :meth:`.RegistryEvents.resolve_type_annotation` event
+
+ :class:`.RegistryEvents` - Complete API reference for all registry events
+
+:ticket:`9832`
+
New Features and Improvements - Core
=====================================
--- /dev/null
+.. change::
+ :tags: feature, orm
+ :tickets: 9832
+
+ Added :class:`_orm.RegistryEvents` event class that allows event listeners
+ to be established on a :class:`_orm.registry` object. The new class
+ provides three events: :meth:`_orm.RegistryEvents.resolve_type_annotation`
+ which allows customization of type annotation resolution that can
+ supplement or replace the use of the
+ :paramref:`.registry.type_annotation_map` dictionary, including that it can
+ be helpful with custom resolution for complex types such as those of
+ :pep:`695`, as well as :meth:`_orm.RegistryEvents.before_configured` and
+ :meth:`_orm.RegistryEvents.after_configured`, which are registry-local
+ forms of the mapper-wide version of these hooks.
+
+ .. seealso::
+
+ :ref:`change_9832`
.. _orm_declarative_mapped_column_type_map_pep593:
-Mapping Multiple Type Configurations to Python Types
-^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+Mapping Multiple Type Configurations to Python Types with pep-593 ``Annotated``
+^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
As individual Python types may be associated with :class:`_types.TypeEngine`
.. _orm_declarative_mapped_column_pep593:
-Mapping Whole Column Declarations to Python Types
-^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+Mapping Whole Column Declarations to Python Types with pep-593 ``Annotated``
+^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
The previous section illustrated using :pep:`593` ``Annotated`` type
:class:`._sqltypes.JSON` instance. Other ``Literal`` variants will continue
to resolve to :class:`_sqltypes.Enum` datatypes.
+.. _orm_declarative_resolve_type_event:
+Resolving Types Programmatically with Events
+^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+
+.. versionadded:: 2.1
+
+The :paramref:`_orm.registry.type_annotation_map` is the usual
+way to customize how :func:`_orm.mapped_column` types are assigned to Python
+types. But for automation of whole classes of types or other custom rules,
+the type map resolution can be augmented and/or replaced using the
+:meth:`.RegistryEvents.resolve_type_annotation` hook.
+
+This event hook allows for dynamic type resolution that goes beyond the static
+mappings possible with :paramref:`_orm.registry.type_annotation_map`. It's
+particularly useful when working with generic types, complex type hierarchies,
+or when you need to implement custom logic for determining SQL types based
+on Python type annotations.
+
+Basic Type Resolution with :meth:`.RegistryEvents.resolve_type_annotation`
+~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+
+Basic type resolution can be set up by registering the event against
+a :class:`_orm.registry` or :class:`_orm.DeclarativeBase` class. The event
+receives a single parameter that allows inspection of the type annotation
+and provides hooks for custom resolution logic.
+
+The following example shows how to use the hook to resolve custom type aliases
+to appropriate SQL types::
+
+ from __future__ import annotations
+
+ from typing import Annotated
+ from typing import Any
+ from typing import get_args
+
+ from sqlalchemy import create_engine
+ from sqlalchemy import event
+ from sqlalchemy import Integer
+ from sqlalchemy import String
+ from sqlalchemy.orm import DeclarativeBase
+ from sqlalchemy.orm import Mapped
+ from sqlalchemy.orm import mapped_column
+ from sqlalchemy.orm import TypeResolve
+ from sqlalchemy.types import TypeEngine
+
+ # Define some custom type aliases
+ type UserId = int
+ type Username = str
+ LongText = Annotated[str, "long"]
+
+
+ class Base(DeclarativeBase):
+ pass
+
+
+ @event.listens_for(Base.registry, "resolve_type_annotation")
+ def resolve_custom_types(resolve_type: TypeResolve) -> TypeEngine[Any] | None:
+ # Handle our custom type aliases
+ if resolve_type.raw_pep_695_type is UserId:
+ return Integer()
+ elif resolve_type.raw_pep_695_type is Username:
+ return String(50)
+ elif resolve_type.raw_pep_593_type:
+ inner_type, *metadata = get_args(resolve_type.raw_pep_593_type)
+ if inner_type is str and "long" in metadata:
+ return String(1000)
+
+ # Fall back to default resolution
+ return None
+
+
+ class User(Base):
+ __tablename__ = "user"
+
+ id: Mapped[UserId] = mapped_column(primary_key=True)
+ name: Mapped[Username]
+ description: Mapped[LongText]
+
+
+ e = create_engine("sqlite://", echo=True)
+ Base.metadata.create_all(e)
+
+In this example, the event handler checks for specific type aliases and
+returns appropriate SQL types. When the handler returns ``None``, the
+default type resolution logic is used.
+
+Programmatic Resolution of pep-695 and NewType types
+~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+
+As detailed in :ref:`orm_declarative_type_map_pep695_types`, SQLAlchemy now
+automatically resolves simple :pep:`695` ``type`` aliases, but does not
+automatically resolve types made using ``typing.NewType`` without
+these types being explicitly present in :paramref:`_orm.registry.type_annotation_map`.
+
+The :meth:`.RegistryEvents.resolve_type_annotation` event provides a way
+to programmatically handle these types. This is particularly useful when you have
+many ``NewType`` instances that would be cumbersome
+to list individually in the type annotation map::
+
+ from __future__ import annotations
+
+ from typing import Annotated
+ from typing import Any
+ from typing import NewType
+
+ from sqlalchemy import event
+ from sqlalchemy import String
+ from sqlalchemy.orm import DeclarativeBase
+ from sqlalchemy.orm import Mapped
+ from sqlalchemy.orm import mapped_column
+ from sqlalchemy.orm import TypeResolve
+ from sqlalchemy.types import TypeEngine
+
+ # Multiple NewType instances
+ IntPK = NewType("IntPK", int)
+ UserId = NewType("UserId", int)
+ ProductId = NewType("ProductId", int)
+ CategoryName = NewType("CategoryName", str)
+
+ # PEP 695 type alias that recursively refers to a NewType
+ type OrderId = Annotated[IntPK, mapped_column(primary_key=True)]
+
+
+ class Base(DeclarativeBase):
+ pass
+
+
+ @event.listens_for(Base.registry, "resolve_type_annotation")
+ def resolve_newtype_and_pep695(resolve_type: TypeResolve) -> TypeEngine[Any] | None:
+
+ # Handle NewType instances by checking their supertype
+ if hasattr(resolve_type.resolved_type, "__supertype__"):
+ supertype = resolve_type.resolved_type.__supertype__
+ if supertype is int:
+ # return default resolution for int
+ return resolve_type.resolve(int)
+ elif supertype is str:
+ return String(100)
+
+ # detect nested pep-695 IntPK type
+ if (
+ resolve_type.resolved_type is IntPK
+ or resolve_type.pep_593_resolved_argument is IntPK
+ ):
+ return resolve_type.resolve(int)
+
+ return None
+
+
+ class Order(Base):
+ __tablename__ = "order"
+
+ id: Mapped[OrderId]
+ user_id: Mapped[UserId]
+ product_id: Mapped[ProductId]
+ category_name: Mapped[CategoryName]
+
+This approach allows you to handle entire categories of types programmatically
+rather than having to enumerate each one in the type annotation map.
+
+
+.. seealso::
+
+ :meth:`.RegistryEvents.resolve_type_annotation`
.. _orm_imperative_table_configuration:
.. autoclass:: sqlalchemy.orm.MapperEvents
:members:
+Registry Events
+---------------
+
+Registry event hooks indicate things happening in reference to a particular
+:class:`_orm.registry`. These include configurational events
+:meth:`_orm.RegistryEvents.before_configured` and
+:meth:`_orm.RegistryEvents.after_configured`, as well as a hook to customize
+type resolution :meth:`_orm.RegistryEvents.resolve_type_annotation`.
+
+.. autoclass:: sqlalchemy.orm.RegistryEvents
+ :members:
+
+.. autoclass:: sqlalchemy.orm.TypeResolve
+ :members:
+
+
Instance Events
---------------
function automatically, against all tables mapped as a subclass
to this class. The function is called via the
``__declare_last__()`` function, which is essentially
- a hook for the :meth:`.after_configured` event.
+ a hook for the :meth:`.MapperEvents.after_configured` event.
:class:`.ConcreteBase` produces a mapped
table for the class itself. Compare to :class:`.AbstractConcreteBase`,
function automatically, against all tables mapped as a subclass
to this class. The function is called via the
``__declare_first__()`` function, which is essentially
- a hook for the :meth:`.before_configured` event.
+ a hook for the :meth:`.MapperEvents.before_configured` event.
:class:`.AbstractConcreteBase` applies :class:`_orm.Mapper` for its
immediately inheriting class, as would occur for any other
from .decl_api import MappedAsDataclass as MappedAsDataclass
from .decl_api import registry as registry
from .decl_api import synonym_for as synonym_for
+from .decl_api import TypeResolve as TypeResolve
from .decl_api import unmapped_dataclass as unmapped_dataclass
from .decl_base import MappedClassProtocol as MappedClassProtocol
from .descriptor_props import Composite as Composite
from .events import InstrumentationEvents as InstrumentationEvents
from .events import MapperEvents as MapperEvents
from .events import QueryEvents as QueryEvents
+from .events import RegistryEvents as RegistryEvents
from .events import SessionEvents as SessionEvents
from .identity import IdentityMap as IdentityMap
from .instrumentation import ClassManager as ClassManager
from .. import exc
from .. import inspection
from .. import util
+from ..event import dispatcher
+from ..event import EventTarget
from ..sql import sqltypes
from ..sql.base import _NoArg
from ..sql.elements import SQLCoreOperations
from ..util import typing as compat_typing
from ..util.typing import CallableReference
from ..util.typing import de_optionalize_union_types
+from ..util.typing import GenericProtocol
from ..util.typing import is_generic
from ..util.typing import is_literal
from ..util.typing import LITERAL_TYPES
from ..util.typing import Self
+from ..util.typing import TypeAliasType
if TYPE_CHECKING:
from ._typing import _O
from .interfaces import MapperProperty
from .state import InstanceState # noqa
from ..sql._typing import _TypeEngineArgument
- from ..sql.type_api import _MatchedOnType
+ from ..util.typing import _MatchedOnType
_T = TypeVar("_T", bound=Any)
)
-class registry:
+class registry(EventTarget):
"""Generalized registry for mapping classes.
The :class:`_orm.registry` serves as the basis for maintaining a collection
_dependents: Set[_RegistryType]
_dependencies: Set[_RegistryType]
_new_mappers: bool
+ dispatch: dispatcher["registry"]
def __init__(
self,
}
)
+ def _resolve_type_with_events(
+ self,
+ cls: Any,
+ key: str,
+ raw_annotation: _MatchedOnType,
+ extracted_type: _MatchedOnType,
+ *,
+ raw_pep_593_type: Optional[GenericProtocol[Any]] = None,
+ pep_593_resolved_argument: Optional[_MatchedOnType] = None,
+ raw_pep_695_type: Optional[TypeAliasType] = None,
+ pep_695_resolved_value: Optional[_MatchedOnType] = None,
+ ) -> Optional[sqltypes.TypeEngine[Any]]:
+ """Resolve type with event support for custom type mapping.
+
+ This method fires the resolve_type_annotation event first to allow
+ custom resolution, then falls back to normal resolution.
+
+ """
+
+ if self.dispatch.resolve_type_annotation:
+ type_resolve = TypeResolve(
+ self,
+ cls,
+ key,
+ raw_annotation,
+ (
+ pep_593_resolved_argument
+ if pep_593_resolved_argument is not None
+ else (
+ pep_695_resolved_value
+ if pep_695_resolved_value is not None
+ else extracted_type
+ )
+ ),
+ raw_pep_593_type,
+ pep_593_resolved_argument,
+ raw_pep_695_type,
+ pep_695_resolved_value,
+ )
+
+ for fn in self.dispatch.resolve_type_annotation:
+ result = fn(type_resolve)
+ if result is not None:
+ return sqltypes.to_instance(result) # type: ignore[no-any-return] # noqa: E501
+
+ if raw_pep_695_type is not None:
+ sqltype = self._resolve_type(raw_pep_695_type)
+ if sqltype is not None:
+ return sqltype
+
+ sqltype = self._resolve_type(extracted_type)
+ if sqltype is not None:
+ return sqltype
+
+ if pep_593_resolved_argument is not None:
+ sqltype = self._resolve_type(pep_593_resolved_argument)
+
+ return sqltype
+
def _resolve_type(
self, python_type: _MatchedOnType
) -> Optional[sqltypes.TypeEngine[Any]]:
_RegistryType = registry # noqa
+class TypeResolve:
+ """Primary argument to the :meth:`.RegistryEvents.resolve_type_annotation`
+ event.
+
+ This object contains all the information needed to resolve a Python
+ type to a SQLAlchemy type. The :attr:`.TypeResolve.resolved_type` is
+ typically the main type that's resolved. To resolve an arbitrary
+ Python type against the current type map, the :meth:`.TypeResolve.resolve`
+ method may be used.
+
+ .. versionadded:: 2.1
+
+ """
+
+ __slots__ = (
+ "registry",
+ "cls",
+ "key",
+ "raw_type",
+ "resolved_type",
+ "raw_pep_593_type",
+ "raw_pep_695_type",
+ "pep_593_resolved_argument",
+ "pep_695_resolved_value",
+ )
+
+ cls: Any
+ "The class being processed during declarative mapping"
+
+ registry: "registry"
+ "The :class:`registry` being used"
+
+ key: str
+ "String name of the ORM mapped attribute being processed"
+
+ raw_type: _MatchedOnType
+ """The type annotation object directly from the attribute's annotations.
+
+ It's recommended to look at :attr:`.TypeResolve.resolved_type` or
+ one of :attr:`.TypeResolve.pep_593_resolved_argument` or
+ :attr:`.TypeResolve.pep_695_resolved_value` rather than the raw type, as
+ the raw type will not be de-optionalized.
+
+ """
+
+ resolved_type: _MatchedOnType
+ """The de-optionalized, "resolved" type after accounting for :pep:`695`
+ and :pep:`593` indirection:
+
+ * If the annotation were a plain Python type or simple alias e.g.
+ ``Mapped[int]``, the resolved_type will be ``int``
+ * If the annotation refers to a :pep:`695` type that references a
+ plain Python type or simple alias, e.g. ``type MyType = int``
+ then ``Mapped[MyType]``, the type will refer to the ``__value__``
+ of the :pep:`695` type, e.g. ``int``, the same as
+ :attr:`.TypeResolve.pep_695_resolved_value`.
+ * If the annotation refers to a :pep:`593` ``Annotated`` object, or
+ a :pep:`695` type alias that in turn refers to a :pep:`593` type,
+ then the type will be the inner type inside of the ``Annotated``,
+ e.g. ``MyType = Annotated[float, mapped_column(...)]`` with
+ ``Mapped[MyType]`` becomes ``float``, the same as
+ :attr:`.TypeResolve.pep_593_resolved_argument`.
+
+ """
+
+ raw_pep_593_type: Optional[GenericProtocol[Any]]
+ """The de-optionalized :pep:`593` type, if the raw type referred to one.
+
+ This would refer to an ``Annotated`` object.
+
+ """
+
+ pep_593_resolved_argument: Optional[_MatchedOnType]
+ """The type extracted from a :pep:`593` ``Annotated`` construct, if the
+ type referred to one.
+
+ When present, this type would be the same as the
+ :attr:`.TypeResolve.resolved_type`.
+
+ """
+
+ raw_pep_695_type: Optional[TypeAliasType]
+ "The de-optionalized :pep:`695` type, if the raw type referred to one."
+
+ pep_695_resolved_value: Optional[_MatchedOnType]
+ """The de-optionalized type referenced by the raw :pep:`695` type, if the
+ raw type referred to one.
+
+ When present, and a :pep:`593` type is not present, this type would be the
+ same as the :attr:`.TypeResolve.resolved_type`.
+
+ """
+
+ def __init__(
+ self,
+ registry: RegistryType,
+ cls: Any,
+ key: str,
+ raw_type: _MatchedOnType,
+ resolved_type: _MatchedOnType,
+ raw_pep_593_type: Optional[GenericProtocol[Any]],
+ pep_593_resolved_argument: Optional[_MatchedOnType],
+ raw_pep_695_type: Optional[TypeAliasType],
+ pep_695_resolved_value: Optional[_MatchedOnType],
+ ):
+ self.registry = registry
+ self.cls = cls
+ self.key = key
+ self.raw_type = raw_type
+ self.resolved_type = resolved_type
+ self.raw_pep_593_type = raw_pep_593_type
+ self.pep_593_resolved_argument = pep_593_resolved_argument
+ self.raw_pep_695_type = raw_pep_695_type
+ self.pep_695_resolved_value = pep_695_resolved_value
+
+ def resolve(
+ self, python_type: _MatchedOnType
+ ) -> Optional[sqltypes.TypeEngine[Any]]:
+ """Resolve the given python type using the type_annotation_map of
+ the :class:`registry`.
+
+ :param python_type: a Python type (e.g. ``int``, ``str``, etc.) Any
+ type object that's present in
+ :paramref:`_orm.registry_type_annotation_map` should produce a
+ non-``None`` result.
+ :return: a SQLAlchemy :class:`.TypeEngine` instance
+ (e.g. :class:`.Integer`,
+ :class:`.String`, etc.), or ``None`` to indicate no type could be
+ matched.
+
+ """
+ return self.registry._resolve_type(python_type)
+
+
def as_declarative(**kw: Any) -> Callable[[Type[_T]], Type[_T]]:
"""
Class decorator which will adapt a given class into a
from typing import Union
import weakref
+from . import decl_api
from . import instrumentation
from . import interfaces
from . import mapperlib
from ..orm.context import QueryContext
from ..orm.decl_api import DeclarativeAttributeIntercept
from ..orm.decl_api import DeclarativeMeta
+ from ..orm.decl_api import registry
from ..orm.mapper import Mapper
from ..orm.state import InstanceState
"event target, use the 'sqlalchemy.orm.Mapper' class.",
"2.0",
)
- return mapperlib.Mapper
+ target = mapperlib.Mapper
+
+ if identifier in ("before_configured", "after_configured"):
+ if target is mapperlib.Mapper:
+ return target
+ else:
+ return None
+
elif isinstance(target, type):
if issubclass(target, mapperlib.Mapper):
return target
event_key._listen_fn,
)
- if (
- identifier in ("before_configured", "after_configured")
- and target is not mapperlib.Mapper
- ):
- util.warn(
- "'before_configured' and 'after_configured' ORM events "
- "only invoke with the Mapper class "
- "as the target."
- )
-
if not raw or not retval:
if not raw:
meth = getattr(cls, identifier)
:meth:`.MapperEvents.after_configured`
+ :meth:`.RegistryEvents.before_configured`
+
+ :meth:`.RegistryEvents.after_configured`
+
:meth:`.MapperEvents.mapper_configured`
"""
:meth:`.MapperEvents.after_configured`
+ :meth:`.RegistryEvents.before_configured`
+
+ :meth:`.RegistryEvents.after_configured`
+
:meth:`.MapperEvents.before_mapper_configured`
"""
:meth:`.MapperEvents.after_configured`
+ :meth:`.RegistryEvents.before_configured`
+
+ :meth:`.RegistryEvents.after_configured`
+
"""
@event._omit_standard_example
:meth:`.MapperEvents.before_configured`
+ :meth:`.RegistryEvents.before_configured`
+
+ :meth:`.RegistryEvents.after_configured`
+
"""
def before_insert(
wrap._bake_ok = bake_ok # type: ignore [attr-defined]
event_key.base_listen(**kw)
+
+
+class RegistryEvents(event.Events["registry"]):
+ """Define events specific to :class:`_orm.registry` lifecycle.
+
+ The :class:`_orm.RegistryEvents` class defines events that are specific
+ to the lifecycle and operation of the :class:`_orm.registry` object.
+
+ e.g.::
+
+ from typing import Any
+
+ from sqlalchemy import event
+ from sqlalchemy.orm import registry
+ from sqlalchemy.orm import TypeResolve
+ from sqlalchemy.types import TypeEngine
+
+ reg = registry()
+
+
+ @event.listens_for(reg, "resolve_type_annotation")
+ def resolve_custom_type(
+ resolve_type: TypeResolve,
+ ) -> TypeEngine[Any] | None:
+ if python_type is MyCustomType:
+ return MyCustomSQLType()
+ return None
+
+ The events defined by :class:`_orm.RegistryEvents` include
+ :meth:`_orm.RegistryEvents.resolve_type_annotation`,
+ :meth:`_orm.RegistryEvents.before_configured`, and
+ :meth:`_orm.RegistryEvents.after_configured`.`. These events may be
+ applied to a :class:`_orm.registry` object as shown in the preceding
+ example, as well as to a declarative base class directly, which will
+ automtically locate the registry for the event to be applied::
+
+ from typing import Any
+
+ from sqlalchemy import event
+ from sqlalchemy.orm import DeclarativeBase
+ from sqlalchemy.orm import registry as RegistryType
+ from sqlalchemy.orm import TypeResolve
+ from sqlalchemy.types import TypeEngine
+
+
+ class Base(DeclarativeBase):
+ pass
+
+
+ @event.listens_for(Base, "resolve_type_annotation")
+ def resolve_custom_type(
+ resolve_type: TypeResolve,
+ ) -> TypeEngine[Any] | None:
+ if resolve_type.resolved_type is MyCustomType:
+ return MyCustomSQLType()
+ else:
+ return None
+
+
+ @event.listens_for(Base, "after_configured")
+ def after_base_configured(registry: RegistryType) -> None:
+ print(f"Registry {registry} fully configured")
+
+ .. versionadded:: 2.1
+
+
+ """
+
+ _target_class_doc = "SomeRegistry"
+ _dispatch_target = decl_api.registry
+
+ @classmethod
+ def _accept_with(
+ cls,
+ target: Any,
+ identifier: str,
+ ) -> Any:
+ if isinstance(target, decl_api.registry):
+ return target
+ elif (
+ isinstance(target, type)
+ and "_sa_registry" in target.__dict__
+ and isinstance(target.__dict__["_sa_registry"], decl_api.registry)
+ ):
+ return target._sa_registry # type: ignore[attr-defined]
+ else:
+ return None
+
+ @classmethod
+ def _listen(
+ cls,
+ event_key: _EventKey["registry"],
+ **kw: Any,
+ ) -> None:
+ identifier = event_key.identifier
+
+ # Only resolve_type_annotation needs retval=True
+ if identifier == "resolve_type_annotation":
+ kw["retval"] = True
+
+ event_key.base_listen(**kw)
+
+ def resolve_type_annotation(
+ self, resolve_type: decl_api.TypeResolve
+ ) -> Optional[Any]:
+ """Intercept and customize type annotation resolution.
+
+ This event is fired when the :class:`_orm.registry` attempts to
+ resolve a Python type annotation to a SQLAlchemy type. This is
+ particularly useful for handling advanced typing scenarios such as
+ nested :pep:`695` type aliases.
+
+ The :meth:`.RegistryEvents.resolve_type_annotation` event automatically
+ sets up ``retval=True`` when the event is set up, so that implementing
+ functions may return a resolved type, or ``None`` to indicate no type
+ was resolved, and the default resolution for the type should proceed.
+
+ :param resolve_type: A :class:`_orm.TypeResolve` object which contains
+ all the relevant information about the type, including a link to the
+ registry and its resolver function.
+
+ :return: A SQLAlchemy type to use for the given Python type. If
+ ``None`` is returned, the default resolution behavior will proceed
+ from there.
+
+ .. versionadded:: 2.1
+
+ .. seealso::
+
+ :ref:`orm_declarative_resolve_type_event`
+
+ """
+
+ def before_configured(self, registry: "registry") -> None:
+ """Called before a series of mappers in this registry are configured.
+
+ This event is invoked each time the :func:`_orm.configure_mappers`
+ function is invoked and this registry has mappers that are part of
+ the configuration process.
+
+ Compared to the :meth:`.MapperEvents.before_configured` event hook,
+ this event is local to the mappers within a specific
+ :class:`_orm.registry` and not for all :class:`.Mapper` objects
+ globally.
+
+ :param registry: The :class:`_orm.registry` instance.
+
+ .. versionadded:: 2.1
+
+ .. seealso::
+
+ :meth:`.RegistryEvents.after_configured`
+
+ :meth:`.MapperEvents.before_configured`
+
+ :meth:`.MapperEvents.after_configured`
+
+ """
+
+ def after_configured(self, registry: "registry") -> None:
+ """Called after a series of mappers in this registry are configured.
+
+ This event is invoked each time the :func:`_orm.configure_mappers`
+ function completes and this registry had mappers that were part of
+ the configuration process.
+
+ Compared to the :meth:`.MapperEvents.after_configured` event hook, this
+ event is local to the mappers within a specific :class:`_orm.registry`
+ and not for all :class:`.Mapper` objects globally.
+
+ :param registry: The :class:`_orm.registry` instance.
+
+ .. versionadded:: 2.1
+
+ .. seealso::
+
+ :meth:`.RegistryEvents.before_configured`
+
+ :meth:`.MapperEvents.before_configured`
+
+ :meth:`.MapperEvents.after_configured`
+
+ """
work; this can be used to establish additional options, properties, or
related mappings before the operation proceeds.
+ * :meth:`.RegistryEvents.before_configured` - Like
+ :meth:`.MapperEvents.before_configured`, but local to a specific
+ :class:`_orm.registry`.
+
+ .. versionadded:: 2.1 - added :meth:`.RegistryEvents.before_configured`
+
* :meth:`.MapperEvents.mapper_configured` - called as each individual
:class:`_orm.Mapper` is configured within the process; will include all
mapper state except for backrefs set up by other mappers that are still
if they are in other :class:`_orm.registry` collections not part of the
current scope of configuration.
+ * :meth:`.RegistryEvents.after_configured` - Like
+ :meth:`.MapperEvents.after_configured`, but local to a specific
+ :class:`_orm.registry`.
+
+ .. versionadded:: 2.1 - added :meth:`.RegistryEvents.after_configured`
+
"""
_configure_registries(_all_registries(), cascade=True)
return
Mapper.dispatch._for_class(Mapper).before_configured() # type: ignore # noqa: E501
+
# initialize properties on all mappers
# note that _mapper_registry is unordered, which
# may randomly conceal/reveal issues related to
# the order of mapper compilation
- _do_configure_registries(registries, cascade)
+ registries_configured = list(
+ _do_configure_registries(registries, cascade)
+ )
+
finally:
_already_compiling = False
+ for reg in registries_configured:
+ reg.dispatch.after_configured(reg)
Mapper.dispatch._for_class(Mapper).after_configured() # type: ignore
@util.preload_module("sqlalchemy.orm.decl_api")
def _do_configure_registries(
registries: Set[_RegistryType], cascade: bool
-) -> None:
+) -> Iterator[registry]:
registry = util.preloaded.orm_decl_api.registry
orig = set(registries)
for reg in registry._recurse_with_dependencies(registries):
+ if reg._new_mappers:
+ reg.dispatch.before_configured(reg)
+
has_skip = False
for mapper in reg._mappers_to_configure():
if not hasattr(exc, "_configure_failed"):
mapper._configure_failed = exc
raise
+
+ if reg._new_mappers:
+ yield reg
if not has_skip:
reg._new_mappers = False
from ..util.typing import Self
if TYPE_CHECKING:
+ from typing import ForwardRef
+
from ._typing import _IdentityKeyType
from ._typing import _InstanceDict
from ._typing import _ORMColumnExprArgument
from ..sql.elements import NamedColumn
from ..sql.operators import OperatorType
from ..util.typing import _AnnotationScanType
+ from ..util.typing import _MatchedOnType
from ..util.typing import RODescriptorReference
_T = TypeVar("_T", bound=Any)
) -> None:
sqltype = self.column.type
+ de_stringified_argument: _MatchedOnType
+
if is_fwd_ref(
argument, check_generic=True, check_for_plain_string=True
):
assert originating_module is not None
- argument = de_stringify_annotation(
+ de_stringified_argument = de_stringify_annotation(
cls, argument, originating_module, include_generic=True
)
+ else:
+ if TYPE_CHECKING:
+ assert not isinstance(argument, (str, ForwardRef))
+ de_stringified_argument = argument
- nullable = includes_none(argument)
+ nullable = includes_none(de_stringified_argument)
if not self._has_nullable:
self.column.nullable = nullable
find_mapped_in: Tuple[Any, ...] = ()
- our_type_is_pep593 = False
- raw_pep_593_type = None
- raw_pep_695_type = None
+ raw_pep_593_type = resolved_pep_593_type = None
+ raw_pep_695_type = resolved_pep_695_type = None
- our_type: Any = de_optionalize_union_types(argument)
+ our_type: Any = de_optionalize_union_types(de_stringified_argument)
if is_pep695(our_type):
raw_pep_695_type = our_type
if our_args:
our_type = our_type[our_args]
- if is_pep593(our_type):
- our_type_is_pep593 = True
+ resolved_pep_695_type = our_type
+ if is_pep593(our_type):
pep_593_components = get_args(our_type)
- raw_pep_593_type = pep_593_components[0]
+ raw_pep_593_type = our_type
+ resolved_pep_593_type = pep_593_components[0]
if nullable:
- raw_pep_593_type = de_optionalize_union_types(raw_pep_593_type)
+ resolved_pep_593_type = de_optionalize_union_types(
+ resolved_pep_593_type
+ )
find_mapped_in = pep_593_components[1:]
use_args_from: Optional[MappedColumn[Any]]
)
if sqltype._isnull and not self.column.foreign_keys:
- checks: List[Any]
- if our_type_is_pep593:
- checks = [our_type, raw_pep_593_type]
- else:
- checks = [our_type]
- if raw_pep_695_type is not None:
- checks.insert(0, raw_pep_695_type)
+ new_sqltype = registry._resolve_type_with_events(
+ cls,
+ key,
+ de_stringified_argument,
+ our_type,
+ raw_pep_593_type=raw_pep_593_type,
+ pep_593_resolved_argument=resolved_pep_593_type,
+ raw_pep_695_type=raw_pep_695_type,
+ pep_695_resolved_value=resolved_pep_695_type,
+ )
- for check_type in checks:
- new_sqltype = registry._resolve_type(check_type)
- if new_sqltype is not None:
- break
- else:
+ if new_sqltype is None:
+ checks = []
+ if raw_pep_695_type:
+ checks.append(raw_pep_695_type)
+ checks.append(our_type)
+ if resolved_pep_593_type:
+ checks.append(resolved_pep_593_type)
if isinstance(our_type, TypeEngine) or (
isinstance(our_type, type)
and issubclass(our_type, TypeEngine)
raise orm_exc.MappedAnnotationError(
f"The object provided inside the {self.column.key!r} "
"attribute Mapped annotation is not a Python type, "
- f"it's the object {argument!r}. Expected a Python "
- "type."
+ f"it's the object {de_stringified_argument!r}. "
+ "Expected a Python type."
)
self.column._set_type(new_sqltype)
from ..util.typing import de_stringify_annotation as _de_stringify_annotation
from ..util.typing import eval_name_only as _eval_name_only
from ..util.typing import fixup_container_fwd_refs
+from ..util.typing import GenericProtocol
from ..util.typing import is_origin_of_cls
from ..util.typing import TupleAny
from ..util.typing import Unpack
from ..sql.selectable import Selectable
from ..sql.visitors import anon_map
from ..util.typing import _AnnotationScanType
+ from ..util.typing import _MatchedOnType
_T = TypeVar("_T", bound=Any)
*,
str_cleanup_fn: Optional[Callable[[str, str], str]] = None,
include_generic: bool = False,
- ) -> Type[Any]: ...
+ ) -> _MatchedOnType: ...
de_stringify_annotation = cast(
else:
return annotated, None
- if len(annotated.__args__) != 1:
+ generic_annotated = cast(GenericProtocol[Any], annotated)
+ if len(generic_annotated.__args__) != 1:
raise orm_exc.MappedAnnotationError(
"Expected sub-type for Mapped[] annotation"
)
return (
# fix dict/list/set args to be ForwardRef, see #11814
- fixup_container_fwd_refs(annotated.__args__[0]),
- annotated.__origin__,
+ fixup_container_fwd_refs(generic_annotated.__args__[0]),
+ generic_annotated.__origin__,
)
from .type_api import _BindProcessorType
from .type_api import _ComparatorFactory
from .type_api import _LiteralProcessorType
- from .type_api import _MatchedOnType
from .type_api import _ResultProcessorType
from ..engine.interfaces import Dialect
+ from ..util.typing import _MatchedOnType
_T = TypeVar("_T", bound="Any")
_CT = TypeVar("_CT", bound=Any)
from typing import Dict
from typing import Generic
from typing import Mapping
-from typing import NewType
from typing import Optional
from typing import overload
from typing import Protocol
from typing import TypeVar
from typing import Union
+from sqlalchemy.util.typing import _MatchedOnType
from .base import SchemaEventTarget
from .cache_key import CacheConst
from .cache_key import NO_CACHE
from .. import exc
from .. import util
from ..util.typing import Self
-from ..util.typing import TypeAliasType
# these are back-assigned by sqltypes.
if typing.TYPE_CHECKING:
from .sqltypes import TABLEVALUE as TABLEVALUE # noqa: F401
from ..engine.interfaces import DBAPIModule
from ..engine.interfaces import Dialect
- from ..util.typing import GenericProtocol
_T = TypeVar("_T", bound=Any)
_T_co = TypeVar("_T_co", bound=Any, covariant=True)
_CT = TypeVar("_CT", bound=Any)
_RT = TypeVar("_RT", bound=Any)
-_MatchedOnType = Union[
- "GenericProtocol[Any]", TypeAliasType, NewType, Type[Any]
-]
-
class _NoValueInList(Enum):
NO_VALUE_IN_LIST = 0
Type[Any], str, ForwardRef, NewType, TypeAliasType, "GenericProtocol[Any]"
]
+_MatchedOnType = Union[
+ "GenericProtocol[Any]", TypeAliasType, NewType, Type[Any]
+]
+
class ArgsTypeProtocol(Protocol):
"""protocol for types that have ``__args__``
return {res}
+@overload
+def is_fwd_ref(
+ type_: _AnnotationScanType,
+ check_generic: bool = ...,
+ check_for_plain_string: Literal[False] = ...,
+) -> TypeGuard[ForwardRef]: ...
+
+
+@overload
+def is_fwd_ref(
+ type_: _AnnotationScanType,
+ check_generic: bool = ...,
+ check_for_plain_string: bool = ...,
+) -> TypeGuard[Union[str, ForwardRef]]: ...
+
+
def is_fwd_ref(
type_: _AnnotationScanType,
check_generic: bool = False,
check_for_plain_string: bool = False,
-) -> TypeGuard[ForwardRef]:
+) -> TypeGuard[Union[str, ForwardRef]]:
if check_for_plain_string and isinstance(type_, str):
return True
elif isinstance(type_, _type_instances.ForwardRef):
def de_optionalize_union_types(type_: Type[Any]) -> Type[Any]: ...
+@overload
+def de_optionalize_union_types(type_: _MatchedOnType) -> _MatchedOnType: ...
+
+
@overload
def de_optionalize_union_types(
type_: _AnnotationScanType,
+import re
+from typing import Annotated
+from typing import Any
+from typing import Optional
+from typing import TypeVar
from unittest.mock import ANY
from unittest.mock import call
from unittest.mock import Mock
from sqlalchemy.orm import class_mapper
from sqlalchemy.orm import configure_mappers
from sqlalchemy.orm import declarative_base
+from sqlalchemy.orm import DeclarativeBase
from sqlalchemy.orm import deferred
from sqlalchemy.orm import EXT_SKIP
from sqlalchemy.orm import instrumentation
from sqlalchemy.orm import joinedload
from sqlalchemy.orm import lazyload
+from sqlalchemy.orm import Mapped
+from sqlalchemy.orm import mapped_column
from sqlalchemy.orm import Mapper
from sqlalchemy.orm import mapperlib
from sqlalchemy.orm import query
+from sqlalchemy.orm import registry
from sqlalchemy.orm import relationship
from sqlalchemy.orm import selectinload
from sqlalchemy.orm import Session
from sqlalchemy.orm import sessionmaker
from sqlalchemy.orm import subqueryload
+from sqlalchemy.orm import TypeResolve
from sqlalchemy.orm import UserDefinedOption
from sqlalchemy.sql.cache_key import NO_CACHE
from sqlalchemy.testing import assert_raises
from sqlalchemy.testing import assert_raises_message
-from sqlalchemy.testing import assert_warns_message
from sqlalchemy.testing import AssertsCompiledSQL
from sqlalchemy.testing import eq_
from sqlalchemy.testing import expect_raises
from sqlalchemy.testing.schema import Column
from sqlalchemy.testing.schema import Table
from sqlalchemy.testing.util import gc_collect
+from sqlalchemy.types import TypeEngine
+from sqlalchemy.util.typing import TypeAliasType
from test.orm import _fixtures
eq_(canary1, ["before_update", "after_update"])
eq_(canary2, [])
- def test_before_after_configured_warn_on_non_mapper(self):
+ @testing.combinations(
+ ("before_configured",), ("after_configured",), argnames="event_name"
+ )
+ @testing.variation(
+ "target_type",
+ [
+ "mappercls",
+ "mapperinstance",
+ "registry",
+ "explicit_base",
+ "imperative_class",
+ "declarative_class",
+ ],
+ )
+ def test_before_after_configured_only_on_mappercls_or_registry(
+ self, event_name, target_type: testing.Variation
+ ):
User, users = self.classes.User, self.tables.users
- m1 = Mock()
+ reg = registry()
- self.mapper_registry.map_imperatively(User, users)
- assert_warns_message(
- sa.exc.SAWarning,
- r"before_configured' and 'after_configured' ORM events only "
- r"invoke with the Mapper class as "
- r"the target.",
- event.listen,
- User,
- "before_configured",
- m1,
+ expect_success = (
+ target_type.mappercls
+ or target_type.registry
+ or target_type.explicit_base
)
- assert_warns_message(
- sa.exc.SAWarning,
- r"before_configured' and 'after_configured' ORM events only "
- r"invoke with the Mapper class as "
- r"the target.",
- event.listen,
- User,
- "after_configured",
- m1,
- )
+ if target_type.mappercls:
+ target = Mapper
+ elif target_type.mapperinstance:
+ reg.map_imperatively(User, users)
+ target = inspect(User)
+ elif target_type.registry:
+ target = reg
+ elif target_type.imperative_class:
+ reg.map_imperatively(User, users)
+ target = User
+ elif target_type.explicit_base:
+
+ class Base(DeclarativeBase):
+ registry = reg
+
+ target = Base
+ elif target_type.declarative_class:
+
+ class Base(DeclarativeBase):
+ registry = reg
+
+ class User(Base):
+ __table__ = users
+
+ target = User
+ else:
+ target_type.fail()
+
+ m1 = Mock()
+ if expect_success:
+ event.listen(target, event_name, m1)
+ else:
+
+ with expect_raises_message(
+ sa_exc.InvalidRequestError,
+ re.escape(
+ f"No such event {event_name!r} for target '{target}'"
+ ),
+ ):
+ event.listen(target, event_name, m1)
def test_before_after_configured(self):
User, users = self.classes.User, self.tables.users
eq_(t1.id, 1)
eq_(t1.prefetch_val, 5)
eq_(t1.returning_val, 5)
+
+
+class RegistryEventsTest(fixtures.MappedTest):
+ """Test RegistryEvents functionality."""
+
+ @testing.variation("scenario", ["direct", "reentrant", "plain"])
+ @testing.variation("include_optional", [True, False])
+ @testing.variation(
+ "type_features",
+ [
+ "none",
+ "plain_pep593",
+ "plain_pep695",
+ "generic_pep593",
+ "plain_pep593_pep695",
+ "generic_pep593_pep695",
+ "generic_pep593_pep695_w_compound",
+ ],
+ )
+ def test_resolve_type_annotation_event(
+ self,
+ scenario: testing.Variation,
+ include_optional: testing.Variation,
+ type_features: testing.Variation,
+ ):
+ reg = registry(type_annotation_map={str: String(70)})
+ Base = reg.generate_base()
+
+ MyCustomType: Any
+ if type_features.none:
+ MyCustomType = type("MyCustomType", (object,), {})
+ elif type_features.plain_pep593:
+ MyCustomType = Annotated[float, mapped_column()]
+ elif type_features.plain_pep695:
+ MyCustomType = TypeAliasType("MyCustomType", float)
+ elif type_features.generic_pep593:
+ T = TypeVar("T")
+ MyCustomType = Annotated[T, mapped_column()]
+ elif type_features.plain_pep593_pep695:
+ MyCustomType = TypeAliasType( # type: ignore
+ "MyCustomType", Annotated[float, mapped_column()]
+ )
+ elif type_features.generic_pep593_pep695:
+ T = TypeVar("T")
+ MyCustomType = TypeAliasType( # type: ignore
+ "MyCustomType", Annotated[T, mapped_column()], type_params=(T,)
+ )
+ elif type_features.generic_pep593_pep695_w_compound:
+ T = TypeVar("T")
+ MyCustomType = TypeAliasType( # type: ignore
+ "MyCustomType",
+ Annotated[T | float, mapped_column()],
+ type_params=(T,),
+ )
+ else:
+ type_features.fail()
+
+ @event.listens_for(reg, "resolve_type_annotation")
+ def resolve_custom_type(
+ type_resolve: TypeResolve,
+ ) -> TypeEngine[Any] | None:
+ assert type_resolve.cls.__name__ == "MyClass"
+
+ if (
+ type_resolve.resolved_type is int
+ and type_resolve.raw_pep_695_type is None
+ and type_resolve.raw_pep_593_type is None
+ ):
+ return None
+
+ if type_features.none:
+ assert type_resolve.resolved_type is MyCustomType
+ elif type_features.plain_pep593:
+ assert type_resolve.resolved_type is float
+ assert type_resolve.raw_pep_593_type is not None
+ assert type_resolve.raw_pep_593_type.__args__[0] is float
+ assert type_resolve.pep_593_resolved_argument is float
+ elif type_features.plain_pep695:
+ assert type_resolve.raw_pep_695_type is MyCustomType
+ assert type_resolve.pep_695_resolved_value is float
+ assert type_resolve.resolved_type is float
+ elif type_features.generic_pep593:
+ assert type_resolve.raw_pep_695_type is None
+ assert type_resolve.pep_593_resolved_argument is str
+ assert type_resolve.resolved_type is str
+ elif type_features.plain_pep593_pep695:
+ assert type_resolve.raw_pep_695_type is not None
+ assert type_resolve.pep_593_resolved_argument is float
+ assert type_resolve.resolved_type is float
+ assert type_resolve.raw_pep_695_type is MyCustomType
+ elif type_features.generic_pep593_pep695:
+ assert type_resolve.raw_pep_695_type is not None
+ assert type_resolve.pep_593_resolved_argument is str
+ elif type_features.generic_pep593_pep695_w_compound:
+ assert type_resolve.raw_pep_695_type is not None
+ assert type_resolve.raw_pep_695_type.__origin__ is MyCustomType
+ assert type_resolve.pep_593_resolved_argument == str | float
+ assert type_resolve.resolved_type == str | float
+ else:
+ type_features.fail()
+
+ if scenario.direct:
+ return String(50)
+ elif scenario.reentrant:
+ return type_resolve.resolve(str)
+ else:
+ scenario.fail()
+
+ use_type_args = (
+ type_features.generic_pep593
+ or type_features.generic_pep593_pep695
+ or type_features.generic_pep593_pep695_w_compound
+ )
+
+ class MyClass(Base):
+ __tablename__ = "mytable"
+ id: Mapped[int] = mapped_column(primary_key=True)
+
+ if include_optional:
+ if scenario.direct or scenario.reentrant:
+ if use_type_args:
+ data: Mapped[Optional[MyCustomType[str]]]
+ else:
+ data: Mapped[Optional[MyCustomType]]
+ else:
+ data: Mapped[Optional[int]]
+ else:
+ if scenario.direct or scenario.reentrant:
+ if use_type_args:
+ data: Mapped[MyCustomType[str]]
+ else:
+ data: Mapped[MyCustomType]
+ else:
+ data: Mapped[int]
+
+ result = MyClass.data.expression.type
+
+ if scenario.direct:
+ assert isinstance(result, String)
+ eq_(result.length, 50)
+ elif scenario.reentrant:
+ assert isinstance(result, String)
+ eq_(result.length, 70)
+ elif scenario.plain:
+ assert isinstance(result, Integer)
+
+ def test_type_resolve_instantiates_type(self, decl_base):
+ MyType = int
+
+ @event.listens_for(decl_base, "resolve_type_annotation")
+ def resolve_custom_type(
+ type_resolve: TypeResolve,
+ ) -> TypeEngine[Any] | None:
+ if type_resolve.resolved_type is MyType:
+ return Integer # <--- note not instantiated
+
+ class User(decl_base):
+ __tablename__ = "user"
+
+ id: Mapped[MyType] = mapped_column(primary_key=True)
+
+ assert isinstance(User.__table__.c.id.type, Integer)
+
+ @testing.variation(
+ "listen_type", ["registry", "generated_base", "explicit_base"]
+ )
+ def test_before_after_configured_events(self, listen_type):
+ """Test the before_configured and after_configured events."""
+ reg = registry()
+
+ if listen_type.generated_base:
+ Base = reg.generate_base()
+ else:
+
+ class Base(DeclarativeBase):
+ registry = reg
+
+ mock = Mock()
+
+ if listen_type.registry:
+
+ @event.listens_for(reg, "before_configured")
+ def before_configured(registry_inst):
+ mock.before_configured(registry_inst)
+
+ @event.listens_for(reg, "after_configured")
+ def after_configured(registry_inst):
+ mock.after_configured(registry_inst)
+
+ else:
+
+ @event.listens_for(Base, "before_configured")
+ def before_configured(registry_inst):
+ mock.before_configured(registry_inst)
+
+ @event.listens_for(Base, "after_configured")
+ def after_configured(registry_inst):
+ mock.after_configured(registry_inst)
+
+ # Create a simple mapped class to trigger configuration
+ class TestClass(Base):
+ __tablename__ = "test_table"
+ id = Column(Integer, primary_key=True)
+
+ # Configure the registry
+ reg.configure()
+
+ # Check that events were fired in the correct order
+ eq_(
+ mock.mock_calls,
+ [call.before_configured(reg), call.after_configured(reg)],
+ )
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
