# orm/attributes.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

# mypy: allow-untyped-defs, allow-untyped-calls

"""Defines instrumentation for class attributes and their interaction

with instances.

This module is usually not directly visible to user applications, but

defines a large part of the ORM's interactivity.

"""

from __future__ import annotations

import dataclasses

import operator

from typing import Any

from typing import Callable

from typing import cast

from typing import ClassVar

from typing import Dict

from typing import Iterable

from typing import List

from typing import Literal

from typing import NamedTuple

from typing import Optional

from typing import overload

from typing import Sequence

from typing import Tuple

from typing import Type

from typing import TYPE_CHECKING

from typing import TypeGuard

from typing import TypeVar

from typing import Union

from . import collections

from . import exc as orm_exc

from . import interfaces

from ._typing import insp_is_aliased_class

from .base import _DeclarativeMapped

from .base import ATTR_EMPTY

from .base import ATTR_WAS_SET

from .base import CALLABLES_OK

from .base import DEFERRED_HISTORY_LOAD

from .base import DONT_SET

from .base import INCLUDE_PENDING_MUTATIONS # noqa

from .base import INIT_OK

from .base import instance_dict as instance_dict

from .base import instance_state as instance_state

from .base import instance_str

from .base import LOAD_AGAINST_COMMITTED

from .base import LoaderCallableStatus

from .base import manager_of_class as manager_of_class

from .base import Mapped as Mapped # noqa

from .base import NEVER_SET # noqa

from .base import NO_AUTOFLUSH

from .base import NO_CHANGE # noqa

from .base import NO_KEY

from .base import NO_RAISE

from .base import NO_VALUE

from .base import NON_PERSISTENT_OK # noqa

from .base import opt_manager_of_class as opt_manager_of_class

from .base import PASSIVE_CLASS_MISMATCH # noqa

from .base import PASSIVE_NO_FETCH

from .base import PASSIVE_NO_FETCH_RELATED # noqa

from .base import PASSIVE_NO_INITIALIZE

from .base import PASSIVE_NO_RESULT

from .base import PASSIVE_OFF

from .base import PASSIVE_ONLY_PERSISTENT

from .base import PASSIVE_RETURN_NO_VALUE

from .base import PassiveFlag

from .base import RELATED_OBJECT_OK # noqa

from .base import SQL_OK # noqa

from .base import SQLORMExpression

from .base import state_str

from .. import event

from .. import exc

from .. import inspection

from .. import util

from ..event import dispatcher

from ..event import EventTarget

from ..sql import base as sql_base

from ..sql import cache_key

from ..sql import coercions

from ..sql import roles

from ..sql import visitors

from ..sql.cache_key import HasCacheKey

from ..sql.visitors import _TraverseInternalsType

from ..sql.visitors import InternalTraversal

from ..util.typing import Self

if TYPE_CHECKING:

from ._typing import _EntityType

from ._typing import _ExternalEntityType

from ._typing import _InstanceDict

from ._typing import _InternalEntityType

from ._typing import _LoaderCallable

from ._typing import _O

from .collections import _AdaptedCollectionProtocol

from .collections import CollectionAdapter

from .interfaces import MapperProperty

from .relationships import RelationshipProperty

from .state import InstanceState

from .util import AliasedInsp

from .writeonly import _WriteOnlyAttributeImpl

from ..event.base import _Dispatch

from ..sql._typing import _ColumnExpressionArgument

from ..sql._typing import _DMLColumnArgument

from ..sql._typing import _InfoType

from ..sql._typing import _PropagateAttrsType

from ..sql.annotation import _AnnotationDict

from ..sql.elements import ColumnElement

from ..sql.elements import Label

from ..sql.operators import OperatorType

from ..sql.selectable import FromClause

_T = TypeVar("_T")

_T_co = TypeVar("_T_co", bound=Any, covariant=True)

_AllPendingType = Sequence[

Tuple[Optional["InstanceState[Any]"], Optional[object]]

]

_UNKNOWN_ATTR_KEY = object()

@inspection._self_inspects

class QueryableAttribute(

_DeclarativeMapped[_T_co],

SQLORMExpression[_T_co],

interfaces.InspectionAttr,

interfaces.PropComparator[_T_co],

roles.JoinTargetRole,

roles.OnClauseRole,

sql_base.Immutable,

cache_key.SlotsMemoizedHasCacheKey,

util.MemoizedSlots,

EventTarget,

):

"""Base class for :term:`descriptor` objects that intercept

attribute events on behalf of a :class:`.MapperProperty`

object. The actual :class:`.MapperProperty` is accessible

via the :attr:`.QueryableAttribute.property`

attribute.

.. seealso::

:class:`.InstrumentedAttribute`

:class:`.MapperProperty`

:attr:`_orm.Mapper.all_orm_descriptors`

:attr:`_orm.Mapper.attrs`

"""

__slots__ = (

"class_",

"key",

"impl",

"comparator",

"property",

"parent",

"expression",

"_of_type",

"_extra_criteria",

"_slots_dispatch",

"_propagate_attrs",

"_doc",

)

is_attribute = True

dispatch: dispatcher[QueryableAttribute[_T_co]]

class_: _ExternalEntityType[Any]

key: str

parententity: _InternalEntityType[Any]

impl: _AttributeImpl

comparator: interfaces.PropComparator[_T_co]

_of_type: Optional[_InternalEntityType[Any]]

_extra_criteria: Tuple[ColumnElement[bool], ...]

_doc: Optional[str]

# PropComparator has a __visit_name__ to participate within

# traversals. Disambiguate the attribute vs. a comparator.

__visit_name__ = "orm_instrumented_attribute"

def __init__(

self,

class_: _ExternalEntityType[_O],

key: str,

parententity: _InternalEntityType[_O],

comparator: interfaces.PropComparator[_T_co],

impl: Optional[_AttributeImpl] = None,

of_type: Optional[_InternalEntityType[Any]] = None,

extra_criteria: Tuple[ColumnElement[bool], ...] = (),

):

self.class_ = class_

self.key = key

self._parententity = self.parent = parententity

# this attribute is non-None after mappers are set up, however in the

# interim class manager setup, there's a check for None to see if it

# needs to be populated, so we assign None here leaving the attribute

# in a temporarily not-type-correct state

self.impl = impl # type: ignore[assignment]

assert comparator is not None

self.comparator = comparator

self._of_type = of_type

self._extra_criteria = extra_criteria

self._doc = None

manager = opt_manager_of_class(class_)

# manager is None in the case of AliasedClass

if manager:

# propagate existing event listeners from

# immediate superclass

for base in manager._bases:

if key in base:

self.dispatch._update(base[key].dispatch)

if base[key].dispatch._active_history:

self.dispatch._active_history = True # type: ignore[attr-defined] # noqa: E501

_cache_key_traversal = [

("key", visitors.ExtendedInternalTraversal.dp_string),

("_parententity", visitors.ExtendedInternalTraversal.dp_multi),

("_of_type", visitors.ExtendedInternalTraversal.dp_multi),

("_extra_criteria", visitors.InternalTraversal.dp_clauseelement_list),

]

def __reduce__(self) -> Any:

# this method is only used in terms of the

# sqlalchemy.ext.serializer extension

return (

_queryable_attribute_unreduce,

(

self.key,

self._parententity.mapper.class_,

self._parententity,

self._parententity.entity,

),

)

@property

def _impl_uses_objects(self) -> bool:

return self.impl.uses_objects

def get_history(

self, instance: Any, passive: PassiveFlag = PASSIVE_OFF

) -> History:

return self.impl.get_history(

instance_state(instance), instance_dict(instance), passive

)

@property

def info(self) -> _InfoType:

"""Return the 'info' dictionary for the underlying SQL element.

The behavior here is as follows:

* If the attribute is a column-mapped property, i.e.

:class:`.ColumnProperty`, which is mapped directly

to a schema-level :class:`_schema.Column` object, this attribute

will return the :attr:`.SchemaItem.info` dictionary associated

with the core-level :class:`_schema.Column` object.

* If the attribute is a :class:`.ColumnProperty` but is mapped to

any other kind of SQL expression other than a

:class:`_schema.Column`,

the attribute will refer to the :attr:`.MapperProperty.info`

dictionary associated directly with the :class:`.ColumnProperty`,

assuming the SQL expression itself does not have its own ``.info``

attribute (which should be the case, unless a user-defined SQL

construct has defined one).

* If the attribute refers to any other kind of

:class:`.MapperProperty`, including :class:`.Relationship`,

the attribute will refer to the :attr:`.MapperProperty.info`

dictionary associated with that :class:`.MapperProperty`.

* To access the :attr:`.MapperProperty.info` dictionary of the

:class:`.MapperProperty` unconditionally, including for a

:class:`.ColumnProperty` that's associated directly with a

:class:`_schema.Column`, the attribute can be referred to using

:attr:`.QueryableAttribute.property` attribute, as

``MyClass.someattribute.property.info``.

.. seealso::

:attr:`.SchemaItem.info`

:attr:`.MapperProperty.info`

"""

return self.comparator.info

parent: _InternalEntityType[Any]

"""Return an inspection instance representing the parent.

This will be either an instance of :class:`_orm.Mapper`

or :class:`.AliasedInsp`, depending upon the nature

of the parent entity which this attribute is associated

with.

"""

expression: ColumnElement[_T_co]

"""The SQL expression object represented by this

:class:`.QueryableAttribute`.

This will typically be an instance of a :class:`_sql.ColumnElement`

subclass representing a column expression.

"""

def _memoized_attr_expression(self) -> ColumnElement[_T]:

annotations: _AnnotationDict

# applies only to Proxy() as used by hybrid.

# currently is an exception to typing rather than feeding through

# non-string keys.

# ideally Proxy() would have a separate set of methods to deal

# with this case.

entity_namespace = self._entity_namespace

assert isinstance(entity_namespace, HasCacheKey)

if self.key is _UNKNOWN_ATTR_KEY:

annotations = {"entity_namespace": entity_namespace}

else:

annotations = {

"proxy_key": self.key,

"proxy_owner": self._parententity,

"entity_namespace": entity_namespace,

}

ce = self.comparator.__clause_element__()

try:

if TYPE_CHECKING:

assert isinstance(ce, ColumnElement)

anno = ce._annotate

except AttributeError as ae:

raise exc.InvalidRequestError(

'When interpreting attribute "%s" as a SQL expression, '

"expected __clause_element__() to return "

"a ClauseElement object, got: %r" % (self, ce)

) from ae

else:

return anno(annotations)

def _memoized_attr__propagate_attrs(self) -> _PropagateAttrsType:

# this suits the case in coercions where we don't actually

# call ``__clause_element__()`` but still need to get

# resolved._propagate_attrs. See #6558.

return util.immutabledict(

{

"compile_state_plugin": "orm",

"plugin_subject": self._parentmapper,

}

)

@property

def _entity_namespace(self) -> _InternalEntityType[Any]:

return self._parententity

@property

def _annotations(self) -> _AnnotationDict:

return self.__clause_element__()._annotations

def __clause_element__(self) -> ColumnElement[_T_co]:

return self.expression

@property

def _from_objects(self) -> List[FromClause]:

return self.expression._from_objects

def _bulk_update_tuples(

self, value: Any

) -> Sequence[Tuple[_DMLColumnArgument, Any]]:

"""Return setter tuples for a bulk UPDATE."""

return self.comparator._bulk_update_tuples(value)

def _bulk_dml_setter(self, key: str) -> Optional[Callable[..., Any]]:

"""return a callable that will process a bulk INSERT value"""

return self.comparator._bulk_dml_setter(key)

def adapt_to_entity(self, adapt_to_entity: AliasedInsp[Any]) -> Self:

assert not self._of_type

return self.__class__(

adapt_to_entity.entity,

self.key,

impl=self.impl,

comparator=self.comparator.adapt_to_entity(adapt_to_entity),

parententity=adapt_to_entity,

)

def of_type(self, entity: _EntityType[_T]) -> QueryableAttribute[_T]:

return QueryableAttribute(

self.class_,

self.key,

self._parententity,

impl=self.impl,

comparator=self.comparator.of_type(entity),

of_type=inspection.inspect(entity),

extra_criteria=self._extra_criteria,

)

def and_(

self, *clauses: _ColumnExpressionArgument[bool]

) -> QueryableAttribute[bool]:

if TYPE_CHECKING:

assert isinstance(self.comparator, RelationshipProperty.Comparator)

exprs = tuple(

coercions.expect(roles.WhereHavingRole, clause)

for clause in util.coerce_generator_arg(clauses)

)

return QueryableAttribute(

self.class_,

self.key,

self._parententity,

impl=self.impl,

comparator=self.comparator.and_(*exprs),

of_type=self._of_type,

extra_criteria=self._extra_criteria + exprs,

)

def _clone(self, **kw: Any) -> QueryableAttribute[_T]:

return QueryableAttribute(

self.class_,

self.key,

self._parententity,

impl=self.impl,

comparator=self.comparator,

of_type=self._of_type,

extra_criteria=self._extra_criteria,

)

def label(self, name: Optional[str]) -> Label[_T_co]:

return self.__clause_element__().label(name)

def operate(

self, op: OperatorType, *other: Any, **kwargs: Any

) -> ColumnElement[Any]:

return op(self.comparator, *other, **kwargs) # type: ignore[no-any-return] # noqa: E501

def reverse_operate(

self, op: OperatorType, other: Any, **kwargs: Any

) -> ColumnElement[Any]:

return op(other, self.comparator, **kwargs) # type: ignore[no-any-return] # noqa: E501

def hasparent(

self, state: InstanceState[Any], optimistic: bool = False

) -> bool:

return self.impl.hasparent(state, optimistic=optimistic) is not False

def _column_strategy_attrs(self) -> Sequence[QueryableAttribute[Any]]:

return (self,)

def __getattr__(self, key: str) -> Any:

try:

return util.MemoizedSlots.__getattr__(self, key)

except AttributeError:

pass

try:

return getattr(self.comparator, key)

except AttributeError as err:

raise AttributeError(

"Neither %r object nor %r object associated with %s "

"has an attribute %r"

% (

type(self).__name__,

type(self.comparator).__name__,

self,

key,

)

) from err

def __str__(self) -> str:

return f"{self.class_.__name__}.{self.key}"

def _memoized_attr_property(self) -> Optional[MapperProperty[Any]]:

return self.comparator.property

def _queryable_attribute_unreduce(

key: str,

mapped_class: Type[_O],

parententity: _InternalEntityType[_O],

entity: _ExternalEntityType[Any],

) -> Any:

# this method is only used in terms of the

# sqlalchemy.ext.serializer extension

if insp_is_aliased_class(parententity):

return entity._get_from_serialized(key, mapped_class, parententity)

else:

return getattr(entity, key)

class InstrumentedAttribute(QueryableAttribute[_T_co]):

"""Class bound instrumented attribute which adds basic

:term:`descriptor` methods.

See :class:`.QueryableAttribute` for a description of most features.

"""

__slots__ = ()

inherit_cache = True

""":meta private:"""

# hack to make __doc__ writeable on instances of

# InstrumentedAttribute, while still keeping classlevel

# __doc__ correct

@util.rw_hybridproperty

def __doc__(self) -> Optional[str]:

return self._doc

@__doc__.setter # type: ignore[no-redef]

def __doc__(self, value: Optional[str]) -> None:

self._doc = value

@__doc__.classlevel # type: ignore[no-redef]

def __doc__(cls) -> Optional[str]:

return super().__doc__

def __set__(self, instance: object, value: Any) -> None:

self.impl.set(

instance_state(instance), instance_dict(instance), value, None

)

def __delete__(self, instance: object) -> None:

self.impl.delete(instance_state(instance), instance_dict(instance))

@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]:

if instance is None:

return self

dict_ = instance_dict(instance)

if self.impl.supports_population and self.key in dict_:

return dict_[self.key] # type: ignore[no-any-return]

else:

try:

state = instance_state(instance)

except AttributeError as err:

raise orm_exc.UnmappedInstanceError(instance) from err

return self.impl.get(state, dict_) # type: ignore[no-any-return]

@dataclasses.dataclass(frozen=True)

class _AdHocHasEntityNamespace(HasCacheKey):

_traverse_internals: ClassVar[_TraverseInternalsType] = [

("_entity_namespace", InternalTraversal.dp_has_cache_key),

]

# py37 compat, no slots=True on dataclass

__slots__ = ("_entity_namespace",)

_entity_namespace: _InternalEntityType[Any]

is_mapper: ClassVar[bool] = False

is_aliased_class: ClassVar[bool] = False

@property

def entity_namespace(self):

return self._entity_namespace.entity_namespace

def _create_proxied_attribute(

descriptor: Any,

) -> Callable[..., QueryableAttribute[Any]]:

"""Create an QueryableAttribute / user descriptor hybrid.

Returns a new QueryableAttribute type that delegates descriptor

behavior and getattr() to the given descriptor.

"""

# TODO: can move this to descriptor_props if the need for this

# function is removed from ext/hybrid.py

class Proxy(QueryableAttribute[_T_co]):

"""Presents the :class:`.QueryableAttribute` interface as a

proxy on top of a Python descriptor / :class:`.PropComparator`

combination.

"""

_extra_criteria = ()

# the attribute error catches inside of __getattr__ basically create a

# singularity if you try putting slots on this too

# __slots__ = ("descriptor", "original_property", "_comparator")

def __init__(

self,

class_: _ExternalEntityType[Any],

key: str,

descriptor: Any,

comparator: interfaces.PropComparator[_T_co],

adapt_to_entity: Optional[AliasedInsp[Any]] = None,

doc: Optional[str] = None,

original_property: Optional[QueryableAttribute[_T_co]] = None,

):

self.class_ = class_

self.key = key

self.descriptor = descriptor

self.original_property = original_property

self._comparator = comparator

self._adapt_to_entity = adapt_to_entity

self._doc = self.__doc__ = doc

@property

def _parententity(self): # type: ignore[override]

return inspection.inspect(self.class_, raiseerr=False)

@property

def parent(self): # type: ignore[override]

return inspection.inspect(self.class_, raiseerr=False)

_is_internal_proxy = True

_cache_key_traversal = [

("key", visitors.ExtendedInternalTraversal.dp_string),

("_parententity", visitors.ExtendedInternalTraversal.dp_multi),

]

def _column_strategy_attrs(self) -> Sequence[QueryableAttribute[Any]]:

prop = self.original_property

if prop is None:

return ()

else:

return prop._column_strategy_attrs()

@property

def _impl_uses_objects(self):

return (

self.original_property is not None

and getattr(self.class_, self.key).impl.uses_objects

)

@property

def _entity_namespace(self):

if hasattr(self._comparator, "_parententity"):

return self._comparator._parententity

else:

# used by hybrid attributes which try to remain

# agnostic of any ORM concepts like mappers

return _AdHocHasEntityNamespace(self._parententity)

@property

def property(self):

return self.comparator.property

@util.memoized_property

def comparator(self):

if callable(self._comparator):

self._comparator = self._comparator()

if self._adapt_to_entity:

self._comparator = self._comparator.adapt_to_entity(

self._adapt_to_entity

)

return self._comparator

def adapt_to_entity(self, adapt_to_entity):

return self.__class__(

adapt_to_entity.entity,

self.key,

self.descriptor,

self._comparator,

adapt_to_entity,

)

def _clone(self, **kw):

return self.__class__(

self.class_,

self.key,

self.descriptor,

self._comparator,

adapt_to_entity=self._adapt_to_entity,

original_property=self.original_property,

)

def __get__(self, instance, owner):

retval = self.descriptor.__get__(instance, owner)

# detect if this is a plain Python @property, which just returns

# itself for class level access. If so, then return us.

# Otherwise, return the object returned by the descriptor.

if retval is self.descriptor and instance is None:

return self

else:

return retval

def __str__(self) -> str:

return f"{self.class_.__name__}.{self.key}"

def __getattr__(self, attribute):

"""Delegate __getattr__ to the original descriptor and/or

comparator."""

# this is unfortunately very complicated, and is easily prone

# to recursion overflows when implementations of related

# __getattr__ schemes are changed

try:

return util.MemoizedSlots.__getattr__(self, attribute)

except AttributeError:

pass

try:

return getattr(descriptor, attribute)

except AttributeError as err:

if attribute == "comparator":

raise AttributeError("comparator") from err

try:

# comparator itself might be unreachable

comparator = self.comparator

except AttributeError as err2:

raise AttributeError(

"Neither %r object nor unconfigured comparator "

"object associated with %s has an attribute %r"

% (type(descriptor).__name__, self, attribute)

) from err2

else:

try:

return getattr(comparator, attribute)

except AttributeError as err3:

raise AttributeError(

"Neither %r object nor %r object "

"associated with %s has an attribute %r"

% (

type(descriptor).__name__,

type(comparator).__name__,

self,

attribute,

)

) from err3

Proxy.__name__ = type(descriptor).__name__ + "Proxy"

util.monkeypatch_proxied_specials(

Proxy, type(descriptor), name="descriptor", from_instance=descriptor

)

return Proxy

OP_REMOVE = util.symbol("REMOVE")

OP_APPEND = util.symbol("APPEND")

OP_REPLACE = util.symbol("REPLACE")

OP_BULK_REPLACE = util.symbol("BULK_REPLACE")

OP_MODIFIED = util.symbol("MODIFIED")

class AttributeEventToken:

"""A token propagated throughout the course of a chain of attribute

events.

Serves as an indicator of the source of the event and also provides

a means of controlling propagation across a chain of attribute

operations.

The :class:`.Event` object is sent as the ``initiator`` argument

when dealing with events such as :meth:`.AttributeEvents.append`,

:meth:`.AttributeEvents.set`,

and :meth:`.AttributeEvents.remove`.

The :class:`.Event` object is currently interpreted by the backref

event handlers, and is used to control the propagation of operations

across two mutually-dependent attributes.

.. versionchanged:: 2.0 Changed the name from ``AttributeEvent``

to ``AttributeEventToken``.

:attribute impl: The :class:`.AttributeImpl` which is the current event

initiator.

:attribute op: The symbol :attr:`.OP_APPEND`, :attr:`.OP_REMOVE`,

:attr:`.OP_REPLACE`, or :attr:`.OP_BULK_REPLACE`, indicating the

source operation.

"""

__slots__ = "impl", "op", "parent_token"

def __init__(self, attribute_impl: _AttributeImpl, op: util.symbol):

self.impl = attribute_impl

self.op = op

self.parent_token = self.impl.parent_token

def __eq__(self, other):

return (

isinstance(other, AttributeEventToken)

and other.impl is self.impl

and other.op == self.op

)

@property

def key(self):

return self.impl.key

def hasparent(self, state):

return self.impl.hasparent(state)

AttributeEvent = AttributeEventToken # legacy

Event = AttributeEventToken # legacy

class _AttributeImpl:

"""internal implementation for instrumented attributes."""

collection: bool

default_accepts_scalar_loader: bool

uses_objects: bool

supports_population: bool

dynamic: bool

_is_has_collection_adapter = False

_replace_token: AttributeEventToken

_remove_token: AttributeEventToken

_append_token: AttributeEventToken

def __init__(

self,

class_: _ExternalEntityType[_O],

key: str,

callable_: Optional[_LoaderCallable],

dispatch: _Dispatch[QueryableAttribute[Any]],

trackparent: bool = False,

compare_function: Optional[Callable[..., bool]] = None,

active_history: bool = False,

parent_token: Optional[AttributeEventToken] = None,

load_on_unexpire: bool = True,

send_modified_events: bool = True,

accepts_scalar_loader: Optional[bool] = None,

**kwargs: Any,

):

r"""Construct an AttributeImpl.

:param \class_: associated class

:param key: string name of the attribute

:param \callable_:

optional function which generates a callable based on a parent

instance, which produces the "default" values for a scalar or

collection attribute when it's first accessed, if not present

already.

:param trackparent:

if True, attempt to track if an instance has a parent attached

to it via this attribute.

:param compare_function:

a function that compares two values which are normally

assignable to this attribute.

:param active_history:

indicates that get_history() should always return the "old" value,

even if it means executing a lazy callable upon attribute change.

:param parent_token:

Usually references the MapperProperty, used as a key for

the hasparent() function to identify an "owning" attribute.

Allows multiple AttributeImpls to all match a single

owner attribute.

:param load_on_unexpire:

if False, don't include this attribute in a load-on-expired

operation, i.e. the "expired_attribute_loader" process.

The attribute can still be in the "expired" list and be

considered to be "expired". Previously, this flag was called

"expire_missing" and is only used by a deferred column

attribute.

:param send_modified_events:

if False, the InstanceState._modified_event method will have no

effect; this means the attribute will never show up as changed in a

history entry.

"""

self.class_ = class_

self.key = key

self.callable_ = callable_

self.dispatch = dispatch

self.trackparent = trackparent

self.parent_token = parent_token or self

self.send_modified_events = send_modified_events

if compare_function is None:

self.is_equal = operator.eq

else:

self.is_equal = compare_function

if accepts_scalar_loader is not None:

self.accepts_scalar_loader = accepts_scalar_loader

else:

self.accepts_scalar_loader = self.default_accepts_scalar_loader

_deferred_history = kwargs.pop("_deferred_history", False)

self._deferred_history = _deferred_history

if active_history:

self.dispatch._active_history = True

self.load_on_unexpire = load_on_unexpire

self._modified_token = AttributeEventToken(self, OP_MODIFIED)

__slots__ = (

"class_",

"key",

"callable_",

"dispatch",

"trackparent",

"parent_token",

"send_modified_events",

"is_equal",

"load_on_unexpire",

"_modified_token",

"accepts_scalar_loader",

"_deferred_history",

)

def __str__(self) -> str:

return f"{self.class_.__name__}.{self.key}"

def _get_active_history(self):

"""Backwards compat for impl.active_history"""

return self.dispatch._active_history

def _set_active_history(self, value):

self.dispatch._active_history = value

active_history = property(_get_active_history, _set_active_history)

def hasparent(

self, state: InstanceState[Any], optimistic: bool = False

) -> bool:

"""Return the boolean value of a `hasparent` flag attached to

the given state.

The `optimistic` flag determines what the default return value

should be if no `hasparent` flag can be located.

As this function is used to determine if an instance is an

*orphan*, instances that were loaded from storage should be

assumed to not be orphans, until a True/False value for this

flag is set.

An instance attribute that is loaded by a callable function

will also not have a `hasparent` flag.

"""

msg = "This AttributeImpl is not configured to track parents."

assert self.trackparent, msg

return (

state.parents.get(id(self.parent_token), optimistic) is not False

)

def sethasparent(

self,

state: InstanceState[Any],

parent_state: InstanceState[Any],

value: bool,

) -> None:

"""Set a boolean flag on the given item corresponding to

whether or not it is attached to a parent object via the

attribute represented by this ``InstrumentedAttribute``.

"""

msg = "This AttributeImpl is not configured to track parents."

assert self.trackparent, msg