Загрузить файлы в «venv/Lib/site-packages/sqlalchemy/orm»

This commit is contained in:
2026-07-02 20:32:06 +00:00
parent 921d6805ec
commit 7597a255b9
5 changed files with 6827 additions and 0 deletions

View File

@@ -0,0 +1,171 @@
# orm/__init__.py
# Copyright (C) 2005-2026 the SQLAlchemy authors and contributors
# <see AUTHORS file>
#
# This module is part of SQLAlchemy and is released under
# the MIT License: https://www.opensource.org/licenses/mit-license.php
"""
Functional constructs for ORM configuration.
See the SQLAlchemy object relational tutorial and mapper configuration
documentation for an overview of how this module is used.
"""
from __future__ import annotations
from typing import Any
from . import exc as exc
from . import mapper as mapperlib
from . import strategy_options as strategy_options
from ._orm_constructors import _mapper_fn as mapper
from ._orm_constructors import aliased as aliased
from ._orm_constructors import backref as backref
from ._orm_constructors import clear_mappers as clear_mappers
from ._orm_constructors import column_property as column_property
from ._orm_constructors import composite as composite
from ._orm_constructors import contains_alias as contains_alias
from ._orm_constructors import create_session as create_session
from ._orm_constructors import deferred as deferred
from ._orm_constructors import dynamic_loader as dynamic_loader
from ._orm_constructors import join as join
from ._orm_constructors import mapped_column as mapped_column
from ._orm_constructors import orm_insert_sentinel as orm_insert_sentinel
from ._orm_constructors import outerjoin as outerjoin
from ._orm_constructors import query_expression as query_expression
from ._orm_constructors import relationship as relationship
from ._orm_constructors import synonym as synonym
from ._orm_constructors import with_loader_criteria as with_loader_criteria
from ._orm_constructors import with_polymorphic as with_polymorphic
from .attributes import AttributeEventToken as AttributeEventToken
from .attributes import InstrumentedAttribute as InstrumentedAttribute
from .attributes import QueryableAttribute as QueryableAttribute
from .base import class_mapper as class_mapper
from .base import DynamicMapped as DynamicMapped
from .base import InspectionAttrExtensionType as InspectionAttrExtensionType
from .base import LoaderCallableStatus as LoaderCallableStatus
from .base import Mapped as Mapped
from .base import NotExtension as NotExtension
from .base import ORMDescriptor as ORMDescriptor
from .base import PassiveFlag as PassiveFlag
from .base import SQLORMExpression as SQLORMExpression
from .base import WriteOnlyMapped as WriteOnlyMapped
from .context import FromStatement as FromStatement
from .context import QueryContext as QueryContext
from .decl_api import add_mapped_attribute as add_mapped_attribute
from .decl_api import as_declarative as as_declarative
from .decl_api import declarative_base as declarative_base
from .decl_api import declarative_mixin as declarative_mixin
from .decl_api import DeclarativeBase as DeclarativeBase
from .decl_api import DeclarativeBaseNoMeta as DeclarativeBaseNoMeta
from .decl_api import DeclarativeMeta as DeclarativeMeta
from .decl_api import declared_attr as declared_attr
from .decl_api import has_inherited_table as has_inherited_table
from .decl_api import mapped_as_dataclass as mapped_as_dataclass
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_base import MappedClassProtocol as MappedClassProtocol
from .descriptor_props import Composite as Composite
from .descriptor_props import CompositeProperty as CompositeProperty
from .descriptor_props import Synonym as Synonym
from .descriptor_props import SynonymProperty as SynonymProperty
from .dynamic import AppenderQuery as AppenderQuery
from .events import AttributeEvents as AttributeEvents
from .events import InstanceEvents as InstanceEvents
from .events import InstrumentationEvents as InstrumentationEvents
from .events import MapperEvents as MapperEvents
from .events import QueryEvents as QueryEvents
from .events import SessionEvents as SessionEvents
from .identity import IdentityMap as IdentityMap
from .instrumentation import ClassManager as ClassManager
from .interfaces import EXT_CONTINUE as EXT_CONTINUE
from .interfaces import EXT_SKIP as EXT_SKIP
from .interfaces import EXT_STOP as EXT_STOP
from .interfaces import InspectionAttr as InspectionAttr
from .interfaces import InspectionAttrInfo as InspectionAttrInfo
from .interfaces import MANYTOMANY as MANYTOMANY
from .interfaces import MANYTOONE as MANYTOONE
from .interfaces import MapperProperty as MapperProperty
from .interfaces import NO_KEY as NO_KEY
from .interfaces import NO_VALUE as NO_VALUE
from .interfaces import ONETOMANY as ONETOMANY
from .interfaces import PropComparator as PropComparator
from .interfaces import RelationshipDirection as RelationshipDirection
from .interfaces import UserDefinedOption as UserDefinedOption
from .loading import merge_frozen_result as merge_frozen_result
from .loading import merge_result as merge_result
from .mapped_collection import attribute_keyed_dict as attribute_keyed_dict
from .mapped_collection import (
attribute_mapped_collection as attribute_mapped_collection,
)
from .mapped_collection import column_keyed_dict as column_keyed_dict
from .mapped_collection import (
column_mapped_collection as column_mapped_collection,
)
from .mapped_collection import keyfunc_mapping as keyfunc_mapping
from .mapped_collection import KeyFuncDict as KeyFuncDict
from .mapped_collection import mapped_collection as mapped_collection
from .mapped_collection import MappedCollection as MappedCollection
from .mapper import configure_mappers as configure_mappers
from .mapper import Mapper as Mapper
from .mapper import reconstructor as reconstructor
from .mapper import validates as validates
from .properties import ColumnProperty as ColumnProperty
from .properties import MappedColumn as MappedColumn
from .properties import MappedSQLExpression as MappedSQLExpression
from .query import AliasOption as AliasOption
from .query import Query as Query
from .relationships import foreign as foreign
from .relationships import Relationship as Relationship
from .relationships import RelationshipProperty as RelationshipProperty
from .relationships import remote as remote
from .scoping import QueryPropertyDescriptor as QueryPropertyDescriptor
from .scoping import scoped_session as scoped_session
from .session import close_all_sessions as close_all_sessions
from .session import make_transient as make_transient
from .session import make_transient_to_detached as make_transient_to_detached
from .session import object_session as object_session
from .session import ORMExecuteState as ORMExecuteState
from .session import Session as Session
from .session import sessionmaker as sessionmaker
from .session import SessionTransaction as SessionTransaction
from .session import SessionTransactionOrigin as SessionTransactionOrigin
from .state import AttributeState as AttributeState
from .state import InstanceState as InstanceState
from .strategy_options import contains_eager as contains_eager
from .strategy_options import defaultload as defaultload
from .strategy_options import defer as defer
from .strategy_options import immediateload as immediateload
from .strategy_options import joinedload as joinedload
from .strategy_options import lazyload as lazyload
from .strategy_options import Load as Load
from .strategy_options import load_only as load_only
from .strategy_options import noload as noload
from .strategy_options import raiseload as raiseload
from .strategy_options import selectin_polymorphic as selectin_polymorphic
from .strategy_options import selectinload as selectinload
from .strategy_options import subqueryload as subqueryload
from .strategy_options import undefer as undefer
from .strategy_options import undefer_group as undefer_group
from .strategy_options import with_expression as with_expression
from .unitofwork import UOWTransaction as UOWTransaction
from .util import Bundle as Bundle
from .util import CascadeOptions as CascadeOptions
from .util import LoaderCriteriaOption as LoaderCriteriaOption
from .util import object_mapper as object_mapper
from .util import polymorphic_union as polymorphic_union
from .util import was_deleted as was_deleted
from .util import with_parent as with_parent
from .writeonly import WriteOnlyCollection as WriteOnlyCollection
from .. import util as _sa_util
def __go(lcls: Any) -> None:
_sa_util.preloaded.import_prefix("sqlalchemy.orm")
_sa_util.preloaded.import_prefix("sqlalchemy.ext")
__go(locals())

File diff suppressed because it is too large Load Diff

View File

@@ -0,0 +1,179 @@
# orm/_typing.py
# Copyright (C) 2022-2026 the SQLAlchemy authors and contributors
# <see AUTHORS file>
#
# This module is part of SQLAlchemy and is released under
# the MIT License: https://www.opensource.org/licenses/mit-license.php
from __future__ import annotations
import operator
from typing import Any
from typing import Dict
from typing import Mapping
from typing import Optional
from typing import Tuple
from typing import Type
from typing import TYPE_CHECKING
from typing import TypeVar
from typing import Union
from ..engine.interfaces import _CoreKnownExecutionOptions
from ..sql import roles
from ..sql._orm_types import DMLStrategyArgument as DMLStrategyArgument
from ..sql._orm_types import (
SynchronizeSessionArgument as SynchronizeSessionArgument,
)
from ..sql._typing import _HasClauseElement
from ..sql.elements import ColumnElement
from ..util.typing import Protocol
from ..util.typing import TypeGuard
if TYPE_CHECKING:
from .attributes import AttributeImpl
from .attributes import CollectionAttributeImpl
from .attributes import HasCollectionAdapter
from .attributes import QueryableAttribute
from .base import PassiveFlag
from .decl_api import registry as _registry_type
from .interfaces import InspectionAttr
from .interfaces import MapperProperty
from .interfaces import ORMOption
from .interfaces import UserDefinedOption
from .mapper import Mapper
from .relationships import RelationshipProperty
from .state import InstanceState
from .util import AliasedClass
from .util import AliasedInsp
from ..sql._typing import _CE
from ..sql.base import ExecutableOption
_T = TypeVar("_T", bound=Any)
_T_co = TypeVar("_T_co", bound=Any, covariant=True)
_O = TypeVar("_O", bound=object)
"""The 'ORM mapped object' type.
"""
if TYPE_CHECKING:
_RegistryType = _registry_type
_InternalEntityType = Union["Mapper[_T]", "AliasedInsp[_T]"]
_ExternalEntityType = Union[Type[_T], "AliasedClass[_T]"]
_EntityType = Union[
Type[_T], "AliasedClass[_T]", "Mapper[_T]", "AliasedInsp[_T]"
]
_ClassDict = Mapping[str, Any]
_InstanceDict = Dict[str, Any]
_IdentityKeyType = Tuple[Type[_T], Tuple[Any, ...], Optional[Any]]
_ORMColumnExprArgument = Union[
ColumnElement[_T],
_HasClauseElement[_T],
roles.ExpressionElementRole[_T],
]
_ORMCOLEXPR = TypeVar("_ORMCOLEXPR", bound=ColumnElement[Any])
class _OrmKnownExecutionOptions(_CoreKnownExecutionOptions, total=False):
populate_existing: bool
autoflush: bool
synchronize_session: SynchronizeSessionArgument
dml_strategy: DMLStrategyArgument
is_delete_using: bool
is_update_from: bool
render_nulls: bool
OrmExecuteOptionsParameter = Union[
_OrmKnownExecutionOptions, Mapping[str, Any]
]
class _ORMAdapterProto(Protocol):
"""protocol for the :class:`.AliasedInsp._orm_adapt_element` method
which is a synonym for :class:`.AliasedInsp._adapt_element`.
"""
def __call__(self, obj: _CE, key: Optional[str] = None) -> _CE: ...
class _LoaderCallable(Protocol):
def __call__(
self, state: InstanceState[Any], passive: PassiveFlag
) -> Any: ...
def is_orm_option(
opt: ExecutableOption,
) -> TypeGuard[ORMOption]:
return not opt._is_core
def is_user_defined_option(
opt: ExecutableOption,
) -> TypeGuard[UserDefinedOption]:
return not opt._is_core and opt._is_user_defined # type: ignore
def is_composite_class(obj: Any) -> bool:
# inlining is_dataclass(obj)
return hasattr(obj, "__composite_values__") or hasattr(
obj, "__dataclass_fields__"
)
if TYPE_CHECKING:
def insp_is_mapper_property(
obj: Any,
) -> TypeGuard[MapperProperty[Any]]: ...
def insp_is_mapper(obj: Any) -> TypeGuard[Mapper[Any]]: ...
def insp_is_aliased_class(obj: Any) -> TypeGuard[AliasedInsp[Any]]: ...
def insp_is_attribute(
obj: InspectionAttr,
) -> TypeGuard[QueryableAttribute[Any]]: ...
def attr_is_internal_proxy(
obj: InspectionAttr,
) -> TypeGuard[QueryableAttribute[Any]]: ...
def prop_is_relationship(
prop: MapperProperty[Any],
) -> TypeGuard[RelationshipProperty[Any]]: ...
def is_collection_impl(
impl: AttributeImpl,
) -> TypeGuard[CollectionAttributeImpl]: ...
def is_has_collection_adapter(
impl: AttributeImpl,
) -> TypeGuard[HasCollectionAdapter]: ...
else:
insp_is_mapper_property = operator.attrgetter("is_property")
insp_is_mapper = operator.attrgetter("is_mapper")
insp_is_aliased_class = operator.attrgetter("is_aliased_class")
insp_is_attribute = operator.attrgetter("is_attribute")
attr_is_internal_proxy = operator.attrgetter("_is_internal_proxy")
is_collection_impl = operator.attrgetter("collection")
prop_is_relationship = operator.attrgetter("_is_relationship")
is_has_collection_adapter = operator.attrgetter(
"_is_has_collection_adapter"
)

File diff suppressed because it is too large Load Diff

View File

@@ -0,0 +1,971 @@
# orm/base.py
# Copyright (C) 2005-2026 the SQLAlchemy authors and contributors
# <see AUTHORS file>
#
# This module is part of SQLAlchemy and is released under
# the MIT License: https://www.opensource.org/licenses/mit-license.php
"""Constants and rudimental functions used throughout the ORM."""
from __future__ import annotations
from enum import Enum
import operator
import typing
from typing import Any
from typing import Callable
from typing import Dict
from typing import Generic
from typing import no_type_check
from typing import Optional
from typing import overload
from typing import Tuple
from typing import Type
from typing import TYPE_CHECKING
from typing import TypeVar
from typing import Union
from . import exc
from ._typing import insp_is_mapper
from .. import exc as sa_exc
from .. import inspection
from .. import util
from ..sql import roles
from ..sql.elements import SQLColumnExpression
from ..sql.elements import SQLCoreOperations
from ..util import FastIntFlag
from ..util.langhelpers import TypingOnly
from ..util.typing import Literal
if typing.TYPE_CHECKING:
from ._typing import _EntityType
from ._typing import _ExternalEntityType
from ._typing import _InternalEntityType
from .attributes import InstrumentedAttribute
from .dynamic import AppenderQuery
from .instrumentation import ClassManager
from .interfaces import PropComparator
from .mapper import Mapper
from .state import InstanceState
from .util import AliasedClass
from .writeonly import WriteOnlyCollection
from ..sql._typing import _ColumnExpressionArgument
from ..sql._typing import _InfoType
from ..sql.elements import ColumnElement
from ..sql.operators import OperatorType
_T = TypeVar("_T", bound=Any)
_T_co = TypeVar("_T_co", bound=Any, covariant=True)
_O = TypeVar("_O", bound=object)
class LoaderCallableStatus(Enum):
PASSIVE_NO_RESULT = 0
"""Symbol returned by a loader callable or other attribute/history
retrieval operation when a value could not be determined, based
on loader callable flags.
"""
PASSIVE_CLASS_MISMATCH = 1
"""Symbol indicating that an object is locally present for a given
primary key identity but it is not of the requested class. The
return value is therefore None and no SQL should be emitted."""
ATTR_WAS_SET = 2
"""Symbol returned by a loader callable to indicate the
retrieved value, or values, were assigned to their attributes
on the target object.
"""
ATTR_EMPTY = 3
"""Symbol used internally to indicate an attribute had no callable."""
NO_VALUE = 4
"""Symbol which may be placed as the 'previous' value of an attribute,
indicating no value was loaded for an attribute when it was modified,
and flags indicated we were not to load it.
"""
NEVER_SET = NO_VALUE
"""
Synonymous with NO_VALUE
.. versionchanged:: 1.4 NEVER_SET was merged with NO_VALUE
"""
(
PASSIVE_NO_RESULT,
PASSIVE_CLASS_MISMATCH,
ATTR_WAS_SET,
ATTR_EMPTY,
NO_VALUE,
) = tuple(LoaderCallableStatus)
NEVER_SET = NO_VALUE
class PassiveFlag(FastIntFlag):
"""Bitflag interface that passes options onto loader callables"""
NO_CHANGE = 0
"""No callables or SQL should be emitted on attribute access
and no state should change
"""
CALLABLES_OK = 1
"""Loader callables can be fired off if a value
is not present.
"""
SQL_OK = 2
"""Loader callables can emit SQL at least on scalar value attributes."""
RELATED_OBJECT_OK = 4
"""Callables can use SQL to load related objects as well
as scalar value attributes.
"""
INIT_OK = 8
"""Attributes should be initialized with a blank
value (None or an empty collection) upon get, if no other
value can be obtained.
"""
NON_PERSISTENT_OK = 16
"""Callables can be emitted if the parent is not persistent."""
LOAD_AGAINST_COMMITTED = 32
"""Callables should use committed values as primary/foreign keys during a
load.
"""
NO_AUTOFLUSH = 64
"""Loader callables should disable autoflush."""
NO_RAISE = 128
"""Loader callables should not raise any assertions"""
DEFERRED_HISTORY_LOAD = 256
"""indicates special load of the previous value of an attribute"""
INCLUDE_PENDING_MUTATIONS = 512
# pre-packaged sets of flags used as inputs
PASSIVE_OFF = (
RELATED_OBJECT_OK | NON_PERSISTENT_OK | INIT_OK | CALLABLES_OK | SQL_OK
)
"Callables can be emitted in all cases."
PASSIVE_RETURN_NO_VALUE = PASSIVE_OFF ^ INIT_OK
"""PASSIVE_OFF ^ INIT_OK"""
PASSIVE_NO_INITIALIZE = PASSIVE_RETURN_NO_VALUE ^ CALLABLES_OK
"PASSIVE_RETURN_NO_VALUE ^ CALLABLES_OK"
PASSIVE_NO_FETCH = PASSIVE_OFF ^ SQL_OK
"PASSIVE_OFF ^ SQL_OK"
PASSIVE_NO_FETCH_RELATED = PASSIVE_OFF ^ RELATED_OBJECT_OK
"PASSIVE_OFF ^ RELATED_OBJECT_OK"
PASSIVE_ONLY_PERSISTENT = PASSIVE_OFF ^ NON_PERSISTENT_OK
"PASSIVE_OFF ^ NON_PERSISTENT_OK"
PASSIVE_MERGE = PASSIVE_OFF | NO_RAISE
"""PASSIVE_OFF | NO_RAISE
Symbol used specifically for session.merge() and similar cases
"""
(
NO_CHANGE,
CALLABLES_OK,
SQL_OK,
RELATED_OBJECT_OK,
INIT_OK,
NON_PERSISTENT_OK,
LOAD_AGAINST_COMMITTED,
NO_AUTOFLUSH,
NO_RAISE,
DEFERRED_HISTORY_LOAD,
INCLUDE_PENDING_MUTATIONS,
PASSIVE_OFF,
PASSIVE_RETURN_NO_VALUE,
PASSIVE_NO_INITIALIZE,
PASSIVE_NO_FETCH,
PASSIVE_NO_FETCH_RELATED,
PASSIVE_ONLY_PERSISTENT,
PASSIVE_MERGE,
) = PassiveFlag.__members__.values()
DEFAULT_MANAGER_ATTR = "_sa_class_manager"
DEFAULT_STATE_ATTR = "_sa_instance_state"
class EventConstants(Enum):
EXT_CONTINUE = 1
EXT_STOP = 2
EXT_SKIP = 3
NO_KEY = 4
"""indicates an :class:`.AttributeEvent` event that did not have any
key argument.
.. versionadded:: 2.0
"""
EXT_CONTINUE, EXT_STOP, EXT_SKIP, NO_KEY = tuple(EventConstants)
class RelationshipDirection(Enum):
"""enumeration which indicates the 'direction' of a
:class:`_orm.RelationshipProperty`.
:class:`.RelationshipDirection` is accessible from the
:attr:`_orm.Relationship.direction` attribute of
:class:`_orm.RelationshipProperty`.
"""
ONETOMANY = 1
"""Indicates the one-to-many direction for a :func:`_orm.relationship`.
This symbol is typically used by the internals but may be exposed within
certain API features.
"""
MANYTOONE = 2
"""Indicates the many-to-one direction for a :func:`_orm.relationship`.
This symbol is typically used by the internals but may be exposed within
certain API features.
"""
MANYTOMANY = 3
"""Indicates the many-to-many direction for a :func:`_orm.relationship`.
This symbol is typically used by the internals but may be exposed within
certain API features.
"""
ONETOMANY, MANYTOONE, MANYTOMANY = tuple(RelationshipDirection)
class InspectionAttrExtensionType(Enum):
"""Symbols indicating the type of extension that a
:class:`.InspectionAttr` is part of."""
class NotExtension(InspectionAttrExtensionType):
NOT_EXTENSION = "not_extension"
"""Symbol indicating an :class:`InspectionAttr` that's
not part of sqlalchemy.ext.
Is assigned to the :attr:`.InspectionAttr.extension_type`
attribute.
"""
_never_set = frozenset([NEVER_SET])
_none_set = frozenset([None, NEVER_SET, PASSIVE_NO_RESULT])
_none_only_set = frozenset([None])
_SET_DEFERRED_EXPIRED = util.symbol("SET_DEFERRED_EXPIRED")
_DEFER_FOR_STATE = util.symbol("DEFER_FOR_STATE")
_RAISE_FOR_STATE = util.symbol("RAISE_FOR_STATE")
_F = TypeVar("_F", bound=Callable[..., Any])
_Self = TypeVar("_Self")
def _assertions(
*assertions: Any,
) -> Callable[[_F], _F]:
@util.decorator
def generate(fn: _F, self: _Self, *args: Any, **kw: Any) -> _Self:
for assertion in assertions:
assertion(self, fn.__name__)
fn(self, *args, **kw)
return self
return generate
if TYPE_CHECKING:
def manager_of_class(cls: Type[_O]) -> ClassManager[_O]: ...
@overload
def opt_manager_of_class(cls: AliasedClass[Any]) -> None: ...
@overload
def opt_manager_of_class(
cls: _ExternalEntityType[_O],
) -> Optional[ClassManager[_O]]: ...
def opt_manager_of_class(
cls: _ExternalEntityType[_O],
) -> Optional[ClassManager[_O]]: ...
def instance_state(instance: _O) -> InstanceState[_O]: ...
def instance_dict(instance: object) -> Dict[str, Any]: ...
else:
# these can be replaced by sqlalchemy.ext.instrumentation
# if augmented class instrumentation is enabled.
def manager_of_class(cls):
try:
return cls.__dict__[DEFAULT_MANAGER_ATTR]
except KeyError as ke:
raise exc.UnmappedClassError(
cls, f"Can't locate an instrumentation manager for class {cls}"
) from ke
def opt_manager_of_class(cls):
return cls.__dict__.get(DEFAULT_MANAGER_ATTR)
instance_state = operator.attrgetter(DEFAULT_STATE_ATTR)
instance_dict = operator.attrgetter("__dict__")
def instance_str(instance: object) -> str:
"""Return a string describing an instance."""
return state_str(instance_state(instance))
def state_str(state: InstanceState[Any]) -> str:
"""Return a string describing an instance via its InstanceState."""
if state is None:
return "None"
else:
return "<%s at 0x%x>" % (state.class_.__name__, id(state.obj()))
def state_class_str(state: InstanceState[Any]) -> str:
"""Return a string describing an instance's class via its
InstanceState.
"""
if state is None:
return "None"
else:
return "<%s>" % (state.class_.__name__,)
def attribute_str(instance: object, attribute: str) -> str:
return instance_str(instance) + "." + attribute
def state_attribute_str(state: InstanceState[Any], attribute: str) -> str:
return state_str(state) + "." + attribute
def object_mapper(instance: _T) -> Mapper[_T]:
"""Given an object, return the primary Mapper associated with the object
instance.
Raises :class:`sqlalchemy.orm.exc.UnmappedInstanceError`
if no mapping is configured.
This function is available via the inspection system as::
inspect(instance).mapper
Using the inspection system will raise
:class:`sqlalchemy.exc.NoInspectionAvailable` if the instance is
not part of a mapping.
"""
return object_state(instance).mapper
def object_state(instance: _T) -> InstanceState[_T]:
"""Given an object, return the :class:`.InstanceState`
associated with the object.
Raises :class:`sqlalchemy.orm.exc.UnmappedInstanceError`
if no mapping is configured.
Equivalent functionality is available via the :func:`_sa.inspect`
function as::
inspect(instance)
Using the inspection system will raise
:class:`sqlalchemy.exc.NoInspectionAvailable` if the instance is
not part of a mapping.
"""
state = _inspect_mapped_object(instance)
if state is None:
raise exc.UnmappedInstanceError(instance)
else:
return state
@inspection._inspects(object)
def _inspect_mapped_object(instance: _T) -> Optional[InstanceState[_T]]:
try:
return instance_state(instance)
except (exc.UnmappedClassError,) + exc.NO_STATE:
return None
def _class_to_mapper(
class_or_mapper: Union[Mapper[_T], Type[_T]],
) -> Mapper[_T]:
# can't get mypy to see an overload for this
insp = inspection.inspect(class_or_mapper, False)
if insp is not None:
return insp.mapper # type: ignore
else:
assert isinstance(class_or_mapper, type)
raise exc.UnmappedClassError(class_or_mapper)
def _mapper_or_none(
entity: Union[Type[_T], _InternalEntityType[_T]],
) -> Optional[Mapper[_T]]:
"""Return the :class:`_orm.Mapper` for the given class or None if the
class is not mapped.
"""
# can't get mypy to see an overload for this
insp = inspection.inspect(entity, False)
if insp is not None:
return insp.mapper # type: ignore
else:
return None
def _is_mapped_class(entity: Any) -> bool:
"""Return True if the given object is a mapped class,
:class:`_orm.Mapper`, or :class:`.AliasedClass`.
"""
insp = inspection.inspect(entity, False)
return (
insp is not None
and not insp.is_clause_element
and (insp.is_mapper or insp.is_aliased_class)
)
def _is_aliased_class(entity: Any) -> bool:
insp = inspection.inspect(entity, False)
return insp is not None and getattr(insp, "is_aliased_class", False)
@no_type_check
def _entity_descriptor(entity: _EntityType[Any], key: str) -> Any:
"""Return a class attribute given an entity and string name.
May return :class:`.InstrumentedAttribute` or user-defined
attribute.
"""
insp = inspection.inspect(entity)
if insp.is_selectable:
description = entity
entity = insp.c
elif insp.is_aliased_class:
entity = insp.entity
description = entity
elif hasattr(insp, "mapper"):
description = entity = insp.mapper.class_
else:
description = entity
try:
return getattr(entity, key)
except AttributeError as err:
raise sa_exc.InvalidRequestError(
"Entity '%s' has no property '%s'" % (description, key)
) from err
if TYPE_CHECKING:
def _state_mapper(state: InstanceState[_O]) -> Mapper[_O]: ...
else:
_state_mapper = util.dottedgetter("manager.mapper")
def _inspect_mapped_class(
class_: Type[_O], configure: bool = False
) -> Optional[Mapper[_O]]:
try:
class_manager = opt_manager_of_class(class_)
if class_manager is None or not class_manager.is_mapped:
return None
mapper = class_manager.mapper
except exc.NO_STATE:
return None
else:
if configure:
mapper._check_configure()
return mapper
def _parse_mapper_argument(arg: Union[Mapper[_O], Type[_O]]) -> Mapper[_O]:
insp = inspection.inspect(arg, raiseerr=False)
if insp_is_mapper(insp):
return insp
raise sa_exc.ArgumentError(f"Mapper or mapped class expected, got {arg!r}")
def class_mapper(class_: Type[_O], configure: bool = True) -> Mapper[_O]:
"""Given a class, return the primary :class:`_orm.Mapper` associated
with the key.
Raises :exc:`.UnmappedClassError` if no mapping is configured
on the given class, or :exc:`.ArgumentError` if a non-class
object is passed.
Equivalent functionality is available via the :func:`_sa.inspect`
function as::
inspect(some_mapped_class)
Using the inspection system will raise
:class:`sqlalchemy.exc.NoInspectionAvailable` if the class is not mapped.
"""
mapper = _inspect_mapped_class(class_, configure=configure)
if mapper is None:
if not isinstance(class_, type):
raise sa_exc.ArgumentError(
"Class object expected, got '%r'." % (class_,)
)
raise exc.UnmappedClassError(class_)
else:
return mapper
class InspectionAttr:
"""A base class applied to all ORM objects and attributes that are
related to things that can be returned by the :func:`_sa.inspect` function.
The attributes defined here allow the usage of simple boolean
checks to test basic facts about the object returned.
While the boolean checks here are basically the same as using
the Python isinstance() function, the flags here can be used without
the need to import all of these classes, and also such that
the SQLAlchemy class system can change while leaving the flags
here intact for forwards-compatibility.
"""
__slots__: Tuple[str, ...] = ()
is_selectable = False
"""Return True if this object is an instance of
:class:`_expression.Selectable`."""
is_aliased_class = False
"""True if this object is an instance of :class:`.AliasedClass`."""
is_instance = False
"""True if this object is an instance of :class:`.InstanceState`."""
is_mapper = False
"""True if this object is an instance of :class:`_orm.Mapper`."""
is_bundle = False
"""True if this object is an instance of :class:`.Bundle`."""
is_property = False
"""True if this object is an instance of :class:`.MapperProperty`."""
is_attribute = False
"""True if this object is a Python :term:`descriptor`.
This can refer to one of many types. Usually a
:class:`.QueryableAttribute` which handles attributes events on behalf
of a :class:`.MapperProperty`. But can also be an extension type
such as :class:`.AssociationProxy` or :class:`.hybrid_property`.
The :attr:`.InspectionAttr.extension_type` will refer to a constant
identifying the specific subtype.
.. seealso::
:attr:`_orm.Mapper.all_orm_descriptors`
"""
_is_internal_proxy = False
"""True if this object is an internal proxy object.
.. versionadded:: 1.2.12
"""
is_clause_element = False
"""True if this object is an instance of
:class:`_expression.ClauseElement`."""
extension_type: InspectionAttrExtensionType = NotExtension.NOT_EXTENSION
"""The extension type, if any.
Defaults to :attr:`.interfaces.NotExtension.NOT_EXTENSION`
.. seealso::
:class:`.HybridExtensionType`
:class:`.AssociationProxyExtensionType`
"""
class InspectionAttrInfo(InspectionAttr):
"""Adds the ``.info`` attribute to :class:`.InspectionAttr`.
The rationale for :class:`.InspectionAttr` vs. :class:`.InspectionAttrInfo`
is that the former is compatible as a mixin for classes that specify
``__slots__``; this is essentially an implementation artifact.
"""
__slots__ = ()
@util.ro_memoized_property
def info(self) -> _InfoType:
"""Info dictionary associated with the object, allowing user-defined
data to be associated with this :class:`.InspectionAttr`.
The dictionary is generated when first accessed. Alternatively,
it can be specified as a constructor argument to the
:func:`.column_property`, :func:`_orm.relationship`, or
:func:`.composite`
functions.
.. seealso::
:attr:`.QueryableAttribute.info`
:attr:`.SchemaItem.info`
"""
return {}
class SQLORMOperations(SQLCoreOperations[_T_co], TypingOnly):
__slots__ = ()
if typing.TYPE_CHECKING:
def of_type(
self, class_: _EntityType[Any]
) -> PropComparator[_T_co]: ...
def and_(
self, *criteria: _ColumnExpressionArgument[bool]
) -> PropComparator[bool]: ...
def any( # noqa: A001
self,
criterion: Optional[_ColumnExpressionArgument[bool]] = None,
**kwargs: Any,
) -> ColumnElement[bool]: ...
def has(
self,
criterion: Optional[_ColumnExpressionArgument[bool]] = None,
**kwargs: Any,
) -> ColumnElement[bool]: ...
class ORMDescriptor(Generic[_T_co], TypingOnly):
"""Represent any Python descriptor that provides a SQL expression
construct at the class level."""
__slots__ = ()
if typing.TYPE_CHECKING:
@overload
def __get__(
self, instance: Any, owner: Literal[None]
) -> ORMDescriptor[_T_co]: ...
@overload
def __get__(
self, instance: Literal[None], owner: Any
) -> SQLCoreOperations[_T_co]: ...
@overload
def __get__(self, instance: object, owner: Any) -> _T_co: ...
def __get__(
self, instance: object, owner: Any
) -> Union[ORMDescriptor[_T_co], SQLCoreOperations[_T_co], _T_co]: ...
class _MappedAnnotationBase(Generic[_T_co], TypingOnly):
"""common class for Mapped and similar ORM container classes.
these are classes that can appear on the left side of an ORM declarative
mapping, containing a mapped class or in some cases a collection
surrounding a mapped class.
"""
__slots__ = ()
class SQLORMExpression(
SQLORMOperations[_T_co], SQLColumnExpression[_T_co], TypingOnly
):
"""A type that may be used to indicate any ORM-level attribute or
object that acts in place of one, in the context of SQL expression
construction.
:class:`.SQLORMExpression` extends from the Core
:class:`.SQLColumnExpression` to add additional SQL methods that are ORM
specific, such as :meth:`.PropComparator.of_type`, and is part of the bases
for :class:`.InstrumentedAttribute`. It may be used in :pep:`484` typing to
indicate arguments or return values that should behave as ORM-level
attribute expressions.
.. versionadded:: 2.0.0b4
"""
__slots__ = ()
class Mapped(
SQLORMExpression[_T_co],
ORMDescriptor[_T_co],
_MappedAnnotationBase[_T_co],
roles.DDLConstraintColumnRole,
):
"""Represent an ORM mapped attribute on a mapped class.
This class represents the complete descriptor interface for any class
attribute that will have been :term:`instrumented` by the ORM
:class:`_orm.Mapper` class. Provides appropriate information to type
checkers such as pylance and mypy so that ORM-mapped attributes
are correctly typed.
The most prominent use of :class:`_orm.Mapped` is in
the :ref:`Declarative Mapping <orm_explicit_declarative_base>` form
of :class:`_orm.Mapper` configuration, where used explicitly it drives
the configuration of ORM attributes such as :func:`_orm.mapped_class`
and :func:`_orm.relationship`.
.. seealso::
:ref:`orm_explicit_declarative_base`
:ref:`orm_declarative_table`
.. tip::
The :class:`_orm.Mapped` class represents attributes that are handled
directly by the :class:`_orm.Mapper` class. It does not include other
Python descriptor classes that are provided as extensions, including
:ref:`hybrids_toplevel` and the :ref:`associationproxy_toplevel`.
While these systems still make use of ORM-specific superclasses
and structures, they are not :term:`instrumented` by the
:class:`_orm.Mapper` and instead provide their own functionality
when they are accessed on a class.
.. versionadded:: 1.4
"""
__slots__ = ()
if typing.TYPE_CHECKING:
@overload
def __get__(
self, instance: None, owner: Any
) -> InstrumentedAttribute[_T_co]: ...
@overload
def __get__(self, instance: object, owner: Any) -> _T_co: ...
def __get__(
self, instance: Optional[object], owner: Any
) -> Union[InstrumentedAttribute[_T_co], _T_co]: ...
@classmethod
def _empty_constructor(cls, arg1: Any) -> Mapped[_T_co]: ...
def __set__(
self, instance: Any, value: Union[SQLCoreOperations[_T_co], _T_co]
) -> None: ...
def __delete__(self, instance: Any) -> None: ...
class _MappedAttribute(Generic[_T_co], TypingOnly):
"""Mixin for attributes which should be replaced by mapper-assigned
attributes.
"""
__slots__ = ()
class _DeclarativeMapped(Mapped[_T_co], _MappedAttribute[_T_co]):
"""Mixin for :class:`.MapperProperty` subclasses that allows them to
be compatible with ORM-annotated declarative mappings.
"""
__slots__ = ()
# MappedSQLExpression, Relationship, Composite etc. dont actually do
# SQL expression behavior. yet there is code that compares them with
# __eq__(), __ne__(), etc. Since #8847 made Mapped even more full
# featured including ColumnOperators, we need to have those methods
# be no-ops for these objects, so return NotImplemented to fall back
# to normal comparison behavior.
def operate(self, op: OperatorType, *other: Any, **kwargs: Any) -> Any:
return NotImplemented
__sa_operate__ = operate
def reverse_operate(
self, op: OperatorType, other: Any, **kwargs: Any
) -> Any:
return NotImplemented
class DynamicMapped(_MappedAnnotationBase[_T_co]):
"""Represent the ORM mapped attribute type for a "dynamic" relationship.
The :class:`_orm.DynamicMapped` type annotation may be used in an
:ref:`Annotated Declarative Table <orm_declarative_mapped_column>` mapping
to indicate that the ``lazy="dynamic"`` loader strategy should be used
for a particular :func:`_orm.relationship`.
.. legacy:: The "dynamic" lazy loader strategy is the legacy form of what
is now the "write_only" strategy described in the section
:ref:`write_only_relationship`.
E.g.::
class User(Base):
__tablename__ = "user"
id: Mapped[int] = mapped_column(primary_key=True)
addresses: DynamicMapped[Address] = relationship(
cascade="all,delete-orphan"
)
See the section :ref:`dynamic_relationship` for background.
.. versionadded:: 2.0
.. seealso::
:ref:`dynamic_relationship` - complete background
:class:`.WriteOnlyMapped` - fully 2.0 style version
"""
__slots__ = ()
if TYPE_CHECKING:
@overload
def __get__(
self, instance: None, owner: Any
) -> InstrumentedAttribute[_T_co]: ...
@overload
def __get__(
self, instance: object, owner: Any
) -> AppenderQuery[_T_co]: ...
def __get__(
self, instance: Optional[object], owner: Any
) -> Union[InstrumentedAttribute[_T_co], AppenderQuery[_T_co]]: ...
def __set__(
self, instance: Any, value: typing.Collection[_T_co]
) -> None: ...
class WriteOnlyMapped(_MappedAnnotationBase[_T_co]):
"""Represent the ORM mapped attribute type for a "write only" relationship.
The :class:`_orm.WriteOnlyMapped` type annotation may be used in an
:ref:`Annotated Declarative Table <orm_declarative_mapped_column>` mapping
to indicate that the ``lazy="write_only"`` loader strategy should be used
for a particular :func:`_orm.relationship`.
E.g.::
class User(Base):
__tablename__ = "user"
id: Mapped[int] = mapped_column(primary_key=True)
addresses: WriteOnlyMapped[Address] = relationship(
cascade="all,delete-orphan"
)
See the section :ref:`write_only_relationship` for background.
.. versionadded:: 2.0
.. seealso::
:ref:`write_only_relationship` - complete background
:class:`.DynamicMapped` - includes legacy :class:`_orm.Query` support
"""
__slots__ = ()
if TYPE_CHECKING:
@overload
def __get__(
self, instance: None, owner: Any
) -> InstrumentedAttribute[_T_co]: ...
@overload
def __get__(
self, instance: object, owner: Any
) -> WriteOnlyCollection[_T_co]: ...
def __get__(
self, instance: Optional[object], owner: Any
) -> Union[
InstrumentedAttribute[_T_co], WriteOnlyCollection[_T_co]
]: ...
def __set__(
self, instance: Any, value: typing.Collection[_T_co]
) -> None: ...