Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/site-packages/sqlalchemy/inspection.py: 74%

1# inspection.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"""The inspection module provides the :func:`_sa.inspect` function, 

9which delivers runtime information about a wide variety 

10of SQLAlchemy objects, both within the Core as well as the 

11ORM. 

12 

13The :func:`_sa.inspect` function is the entry point to SQLAlchemy's 

14public API for viewing the configuration and construction 

15of in-memory objects. Depending on the type of object 

16passed to :func:`_sa.inspect`, the return value will either be 

17a related object which provides a known interface, or in many 

18cases it will return the object itself. 

19 

20The rationale for :func:`_sa.inspect` is twofold. One is that 

21it replaces the need to be aware of a large variety of "information 

22getting" functions in SQLAlchemy, such as 

23:meth:`_reflection.Inspector.from_engine` (deprecated in 1.4), 

24:func:`.orm.attributes.instance_state`, :func:`_orm.class_mapper`, 

25and others. The other is that the return value of :func:`_sa.inspect` 

26is guaranteed to obey a documented API, thus allowing third party 

27tools which build on top of SQLAlchemy configurations to be constructed 

28in a forwards-compatible way. 

29 

30""" 

31from __future__ import annotations 

32 

33from typing import Any 

34from typing import Callable 

35from typing import Dict 

36from typing import Generic 

37from typing import Optional 

38from typing import overload 

39from typing import Protocol 

40from typing import Type 

41from typing import TypeVar 

42from typing import Union 

43 

44from . import exc 

45from .util.typing import Literal 

46 

47_T = TypeVar("_T", bound=Any) 

48_TCov = TypeVar("_TCov", bound=Any, covariant=True) 

49_F = TypeVar("_F", bound=Callable[..., Any]) 

50 

51_IN = TypeVar("_IN", bound=Any) 

52 

53_registrars: Dict[type, Union[Literal[True], Callable[[Any], Any]]] = {} 

54 

55 

56class Inspectable(Generic[_T]): 

57 """define a class as inspectable. 

58 

59 This allows typing to set up a linkage between an object that 

60 can be inspected and the type of inspection it returns. 

61 

62 Unfortunately we cannot at the moment get all classes that are 

63 returned by inspection to suit this interface as we get into 

64 MRO issues. 

65 

66 """ 

67 

68 __slots__ = () 

69 

70 

71class _InspectableTypeProtocol(Protocol[_TCov]): 

72 """a protocol defining a method that's used when a type (ie the class 

73 itself) is passed to inspect(). 

74 

75 """ 

76 

77 def _sa_inspect_type(self) -> _TCov: ... 

78 

79 

80class _InspectableProtocol(Protocol[_TCov]): 

81 """a protocol defining a method that's used when an instance is 

82 passed to inspect(). 

83 

84 """ 

85 

86 def _sa_inspect_instance(self) -> _TCov: ... 

87 

88 

89@overload 

90def inspect( 

91 subject: Type[_InspectableTypeProtocol[_IN]], raiseerr: bool = True 

92) -> _IN: ... 

93 

94 

95@overload 

96def inspect( 

97 subject: _InspectableProtocol[_IN], raiseerr: bool = True 

98) -> _IN: ... 

99 

100 

101@overload 

102def inspect(subject: Inspectable[_IN], raiseerr: bool = True) -> _IN: ... 

103 

104 

105@overload 

106def inspect(subject: Any, raiseerr: Literal[False] = ...) -> Optional[Any]: ... 

107 

108 

109@overload 

110def inspect(subject: Any, raiseerr: bool = True) -> Any: ... 

111 

112 

113def inspect(subject: Any, raiseerr: bool = True) -> Any: 

114 """Produce an inspection object for the given target. 

115 

116 The returned value in some cases may be the 

117 same object as the one given, such as if a 

118 :class:`_orm.Mapper` object is passed. In other 

119 cases, it will be an instance of the registered 

120 inspection type for the given object, such as 

121 if an :class:`_engine.Engine` is passed, an 

122 :class:`_reflection.Inspector` object is returned. 

123 

124 :param subject: the subject to be inspected. 

125 :param raiseerr: When ``True``, if the given subject 

126 does not 

127 correspond to a known SQLAlchemy inspected type, 

128 :class:`sqlalchemy.exc.NoInspectionAvailable` 

129 is raised. If ``False``, ``None`` is returned. 

130 

131 """ 

132 type_ = type(subject) 

133 for cls in type_.__mro__: 

134 if cls in _registrars: 

135 reg = _registrars.get(cls, None) 

136 if reg is None: 

137 continue 

138 elif reg is True: 

139 return subject 

140 ret = reg(subject) 

141 if ret is not None: 

142 return ret 

143 else: 

144 reg = ret = None 

145 

146 if raiseerr and (reg is None or ret is None): 

147 raise exc.NoInspectionAvailable( 

148 "No inspection system is " 

149 "available for object of type %s" % type_ 

150 ) 

151 return ret 

152 

153 

154def _inspects( 

155 *types: Type[Any], 

156) -> Callable[[_F], _F]: 

157 def decorate(fn_or_cls: _F) -> _F: 

158 for type_ in types: 

159 if type_ in _registrars: 

160 raise AssertionError("Type %s is already registered" % type_) 

161 _registrars[type_] = fn_or_cls 

162 return fn_or_cls 

163 

164 return decorate 

165 

166 

167_TT = TypeVar("_TT", bound="Type[Any]") 

168 

169 

170def _self_inspects(cls: _TT) -> _TT: 

171 if cls in _registrars: 

172 raise AssertionError("Type %s is already registered" % cls) 

173 _registrars[cls] = True 

174 return cls