codekingpro/portable-devtools

1# orm/exc.py
2# Copyright (C) 2005-2026 the SQLAlchemy authors and contributors
3# <see AUTHORS file>
4#
5# This module is part of SQLAlchemy and is released under
6# the MIT License: https://www.opensource.org/licenses/mit-license.php
7
8"""SQLAlchemy ORM exceptions."""
9
10from __future__ import annotations
11
12from typing import Any
13from typing import Optional
14from typing import Tuple
15from typing import Type
16from typing import TYPE_CHECKING
17from typing import TypeVar
18
19from .util import _mapper_property_as_plain_name
20from .. import exc as sa_exc
21from .. import util
22from ..exc import MultipleResultsFound  # noqa
23from ..exc import NoResultFound  # noqa
24
25if TYPE_CHECKING:
26    from .interfaces import LoaderStrategy
27    from .interfaces import MapperProperty
28    from .state import InstanceState
29
30_T = TypeVar("_T", bound=Any)
31
32NO_STATE = (AttributeError, KeyError)
33"""Exception types that may be raised by instrumentation implementations."""
34
35
36class StaleDataError(sa_exc.SQLAlchemyError):
37    """An operation encountered database state that is unaccounted for.
38
39    Conditions which cause this to happen include:
40
41    * A flush may have attempted to update or delete rows
42      and an unexpected number of rows were matched during
43      the UPDATE or DELETE statement.   Note that when
44      version_id_col is used, rows in UPDATE or DELETE statements
45      are also matched against the current known version
46      identifier.
47
48    * A mapped object with version_id_col was refreshed,
49      and the version number coming back from the database does
50      not match that of the object itself.
51
52    * A object is detached from its parent object, however
53      the object was previously attached to a different parent
54      identity which was garbage collected, and a decision
55      cannot be made if the new parent was really the most
56      recent "parent".
57
58    """
59
60
61ConcurrentModificationError = StaleDataError
62
63
64class FlushError(sa_exc.SQLAlchemyError):
65    """A invalid condition was detected during flush()."""
66
67
68class MappedAnnotationError(sa_exc.ArgumentError):
69    """Raised when ORM annotated declarative cannot interpret the
70    expression present inside of the :class:`.Mapped` construct.
71
72    .. versionadded:: 2.0.40
73
74    """
75
76
77class UnmappedError(sa_exc.InvalidRequestError):
78    """Base for exceptions that involve expected mappings not present."""
79
80
81class ObjectDereferencedError(sa_exc.SQLAlchemyError):
82    """An operation cannot complete due to an object being garbage
83    collected.
84
85    """
86
87
88class DetachedInstanceError(sa_exc.SQLAlchemyError):
89    """An attempt to access unloaded attributes on a
90    mapped instance that is detached."""
91
92    code = "bhk3"
93
94
95class UnmappedInstanceError(UnmappedError):
96    """An mapping operation was requested for an unknown instance."""
97
98    @util.preload_module("sqlalchemy.orm.base")
99    def __init__(self, obj: object, msg: Optional[str] = None):
100        base = util.preloaded.orm_base
101
102        if not msg:
103            try:
104                base.class_mapper(type(obj))
105                name = _safe_cls_name(type(obj))
106                msg = (
107                    "Class %r is mapped, but this instance lacks "
108                    "instrumentation.  This occurs when the instance "
109                    "is created before sqlalchemy.orm.mapper(%s) "
110                    "was called." % (name, name)
111                )
112            except UnmappedClassError:
113                msg = f"Class '{_safe_cls_name(type(obj))}' is not mapped"
114                if isinstance(obj, type):
115                    msg += (
116                        "; was a class (%s) supplied where an instance was "
117                        "required?" % _safe_cls_name(obj)
118                    )
119        UnmappedError.__init__(self, msg)
120
121    def __reduce__(self) -> Any:
122        return self.__class__, (None, self.args[0])
123
124
125class UnmappedClassError(UnmappedError):
126    """An mapping operation was requested for an unknown class."""
127
128    def __init__(self, cls: Type[_T], msg: Optional[str] = None):
129        if not msg:
130            msg = _default_unmapped(cls)
131        UnmappedError.__init__(self, msg)
132
133    def __reduce__(self) -> Any:
134        return self.__class__, (None, self.args[0])
135
136
137class ObjectDeletedError(sa_exc.InvalidRequestError):
138    """A refresh operation failed to retrieve the database
139    row corresponding to an object's known primary key identity.
140
141    A refresh operation proceeds when an expired attribute is
142    accessed on an object, or when :meth:`_query.Query.get` is
143    used to retrieve an object which is, upon retrieval, detected
144    as expired.   A SELECT is emitted for the target row
145    based on primary key; if no row is returned, this
146    exception is raised.
147
148    The true meaning of this exception is simply that
149    no row exists for the primary key identifier associated
150    with a persistent object.   The row may have been
151    deleted, or in some cases the primary key updated
152    to a new value, outside of the ORM's management of the target
153    object.
154
155    """
156
157    @util.preload_module("sqlalchemy.orm.base")
158    def __init__(self, state: InstanceState[Any], msg: Optional[str] = None):
159        base = util.preloaded.orm_base
160
161        if not msg:
162            msg = (
163                "Instance '%s' has been deleted, or its "
164                "row is otherwise not present." % base.state_str(state)
165            )
166
167        sa_exc.InvalidRequestError.__init__(self, msg)
168
169    def __reduce__(self) -> Any:
170        return self.__class__, (None, self.args[0])
171
172
173class UnmappedColumnError(sa_exc.InvalidRequestError):
174    """Mapping operation was requested on an unknown column."""
175
176
177class LoaderStrategyException(sa_exc.InvalidRequestError):
178    """A loader strategy for an attribute does not exist."""
179
180    def __init__(
181        self,
182        applied_to_property_type: Type[Any],
183        requesting_property: MapperProperty[Any],
184        applies_to: Optional[Type[MapperProperty[Any]]],
185        actual_strategy_type: Optional[Type[LoaderStrategy]],
186        strategy_key: Tuple[Any, ...],
187    ):
188        if actual_strategy_type is None:
189            sa_exc.InvalidRequestError.__init__(
190                self,
191                "Can't find strategy %s for %s"
192                % (strategy_key, requesting_property),
193            )
194        else:
195            assert applies_to is not None
196            sa_exc.InvalidRequestError.__init__(
197                self,
198                'Can\'t apply "%s" strategy to property "%s", '
199                'which is a "%s"; this loader strategy is intended '
200                'to be used with a "%s".'
201                % (
202                    util.clsname_as_plain_name(actual_strategy_type),
203                    requesting_property,
204                    _mapper_property_as_plain_name(applied_to_property_type),
205                    _mapper_property_as_plain_name(applies_to),
206                ),
207            )
208
209
210def _safe_cls_name(cls: Type[Any]) -> str:
211    cls_name: Optional[str]
212    try:
213        cls_name = ".".join((cls.__module__, cls.__name__))
214    except AttributeError:
215        cls_name = getattr(cls, "__name__", None)
216        if cls_name is None:
217            cls_name = repr(cls)
218    return cls_name
219
220
221@util.preload_module("sqlalchemy.orm.base")
222def _default_unmapped(cls: Type[Any]) -> Optional[str]:
223    base = util.preloaded.orm_base
224
225    try:
226        mappers = base.manager_of_class(cls).mappers  # type: ignore
227    except (
228        UnmappedClassError,
229        TypeError,
230    ) + NO_STATE:
231        mappers = {}
232    name = _safe_cls_name(cls)
233
234    if not mappers:
235        return f"Class '{name}' is not mapped"
236    else:
237        return None
238