codekingpro/portable-devtools

1# orm/exc.py
2# Copyright (C) 2005-2024 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 UnmappedError(sa_exc.InvalidRequestError):
69    """Base for exceptions that involve expected mappings not present."""
70
71
72class ObjectDereferencedError(sa_exc.SQLAlchemyError):
73    """An operation cannot complete due to an object being garbage
74    collected.
75
76    """
77
78
79class DetachedInstanceError(sa_exc.SQLAlchemyError):
80    """An attempt to access unloaded attributes on a
81    mapped instance that is detached."""
82
83    code = "bhk3"
84
85
86class UnmappedInstanceError(UnmappedError):
87    """An mapping operation was requested for an unknown instance."""
88
89    @util.preload_module("sqlalchemy.orm.base")
90    def __init__(self, obj: object, msg: Optional[str] = None):
91        base = util.preloaded.orm_base
92
93        if not msg:
94            try:
95                base.class_mapper(type(obj))
96                name = _safe_cls_name(type(obj))
97                msg = (
98                    "Class %r is mapped, but this instance lacks "
99                    "instrumentation.  This occurs when the instance "
100                    "is created before sqlalchemy.orm.mapper(%s) "
101                    "was called." % (name, name)
102                )
103            except UnmappedClassError:
104                msg = f"Class '{_safe_cls_name(type(obj))}' is not mapped"
105                if isinstance(obj, type):
106                    msg += (
107                        "; was a class (%s) supplied where an instance was "
108                        "required?" % _safe_cls_name(obj)
109                    )
110        UnmappedError.__init__(self, msg)
111
112    def __reduce__(self) -> Any:
113        return self.__class__, (None, self.args[0])
114
115
116class UnmappedClassError(UnmappedError):
117    """An mapping operation was requested for an unknown class."""
118
119    def __init__(self, cls: Type[_T], msg: Optional[str] = None):
120        if not msg:
121            msg = _default_unmapped(cls)
122        UnmappedError.__init__(self, msg)
123
124    def __reduce__(self) -> Any:
125        return self.__class__, (None, self.args[0])
126
127
128class ObjectDeletedError(sa_exc.InvalidRequestError):
129    """A refresh operation failed to retrieve the database
130    row corresponding to an object's known primary key identity.
131
132    A refresh operation proceeds when an expired attribute is
133    accessed on an object, or when :meth:`_query.Query.get` is
134    used to retrieve an object which is, upon retrieval, detected
135    as expired.   A SELECT is emitted for the target row
136    based on primary key; if no row is returned, this
137    exception is raised.
138
139    The true meaning of this exception is simply that
140    no row exists for the primary key identifier associated
141    with a persistent object.   The row may have been
142    deleted, or in some cases the primary key updated
143    to a new value, outside of the ORM's management of the target
144    object.
145
146    """
147
148    @util.preload_module("sqlalchemy.orm.base")
149    def __init__(self, state: InstanceState[Any], msg: Optional[str] = None):
150        base = util.preloaded.orm_base
151
152        if not msg:
153            msg = (
154                "Instance '%s' has been deleted, or its "
155                "row is otherwise not present." % base.state_str(state)
156            )
157
158        sa_exc.InvalidRequestError.__init__(self, msg)
159
160    def __reduce__(self) -> Any:
161        return self.__class__, (None, self.args[0])
162
163
164class UnmappedColumnError(sa_exc.InvalidRequestError):
165    """Mapping operation was requested on an unknown column."""
166
167
168class LoaderStrategyException(sa_exc.InvalidRequestError):
169    """A loader strategy for an attribute does not exist."""
170
171    def __init__(
172        self,
173        applied_to_property_type: Type[Any],
174        requesting_property: MapperProperty[Any],
175        applies_to: Optional[Type[MapperProperty[Any]]],
176        actual_strategy_type: Optional[Type[LoaderStrategy]],
177        strategy_key: Tuple[Any, ...],
178    ):
179        if actual_strategy_type is None:
180            sa_exc.InvalidRequestError.__init__(
181                self,
182                "Can't find strategy %s for %s"
183                % (strategy_key, requesting_property),
184            )
185        else:
186            assert applies_to is not None
187            sa_exc.InvalidRequestError.__init__(
188                self,
189                'Can\'t apply "%s" strategy to property "%s", '
190                'which is a "%s"; this loader strategy is intended '
191                'to be used with a "%s".'
192                % (
193                    util.clsname_as_plain_name(actual_strategy_type),
194                    requesting_property,
195                    _mapper_property_as_plain_name(applied_to_property_type),
196                    _mapper_property_as_plain_name(applies_to),
197                ),
198            )
199
200
201def _safe_cls_name(cls: Type[Any]) -> str:
202    cls_name: Optional[str]
203    try:
204        cls_name = ".".join((cls.__module__, cls.__name__))
205    except AttributeError:
206        cls_name = getattr(cls, "__name__", None)
207        if cls_name is None:
208            cls_name = repr(cls)
209    return cls_name
210
211
212@util.preload_module("sqlalchemy.orm.base")
213def _default_unmapped(cls: Type[Any]) -> Optional[str]:
214    base = util.preloaded.orm_base
215
216    try:
217        mappers = base.manager_of_class(cls).mappers  # type: ignore
218    except (
219        UnmappedClassError,
220        TypeError,
221    ) + NO_STATE:
222        mappers = {}
223    name = _safe_cls_name(cls)
224
225    if not mappers:
226        return f"Class '{name}' is not mapped"
227    else:
228        return None
229