Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/sqlalchemy/orm/exc.py: 48%

1# orm/exc.py 

2# Copyright (C) 2005-2025 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