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

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