Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/sqlalchemy/util/compat.py: 72%

1# util/compat.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# mypy: allow-untyped-defs, allow-untyped-calls 

8 

9"""Handle Python version/platform incompatibilities.""" 

10 

11from __future__ import annotations 

12 

13import base64 

14import dataclasses 

15import hashlib 

16from importlib import metadata as importlib_metadata 

17import inspect 

18import operator 

19import platform 

20import sys 

21import sysconfig 

22import typing 

23from typing import Any 

24from typing import Callable 

25from typing import Dict 

26from typing import Iterable 

27from typing import List 

28from typing import Mapping 

29from typing import Optional 

30from typing import Sequence 

31from typing import Set 

32from typing import Tuple 

33from typing import Type 

34 

35py314b1 = sys.version_info >= (3, 14, 0, "beta", 1) 

36py314 = sys.version_info >= (3, 14) 

37py313 = sys.version_info >= (3, 13) 

38py312 = sys.version_info >= (3, 12) 

39pypy = platform.python_implementation() == "PyPy" 

40cpython = platform.python_implementation() == "CPython" 

41freethreading = bool(sysconfig.get_config_var("Py_GIL_DISABLED")) 

42 

43win32 = sys.platform.startswith("win") 

44osx = sys.platform.startswith("darwin") 

45arm = "aarch" in platform.machine().lower() 

46is64bit = sys.maxsize > 2**32 

47 

48has_refcount_gc = bool(cpython) 

49 

50dottedgetter = operator.attrgetter 

51 

52 

53# use sys.version_info to enable mypy version narrowing 

54if sys.version_info >= (3, 14): 

55 

56 import annotationlib 

57 from string.templatelib import Template as Template 

58 

59 def get_annotations(obj: Any) -> Mapping[str, Any]: 

60 return annotationlib.get_annotations( 

61 obj, format=annotationlib.Format.FORWARDREF 

62 ) 

63 

64else: 

65 

66 def get_annotations(obj: Any) -> Mapping[str, Any]: 

67 return inspect.get_annotations(obj) 

68 

69 class Template: 

70 """Minimal Template for Python < 3.14 (test usage only).""" 

71 

72 def __init__(self, *parts: Any): 

73 self._parts = parts 

74 

75 @property 

76 def strings(self) -> Tuple[str, ...]: 

77 return tuple(p for p in self._parts if isinstance(p, str)) 

78 

79 @property 

80 def interpolations(self) -> Tuple[Any, ...]: 

81 return tuple(p for p in self._parts if not isinstance(p, str)) 

82 

83 def __iter__(self) -> Any: 

84 return iter(self._parts) 

85 

86 

87class FullArgSpec(typing.NamedTuple): 

88 args: List[str] 

89 varargs: Optional[str] 

90 varkw: Optional[str] 

91 defaults: Optional[Tuple[Any, ...]] 

92 kwonlyargs: List[str] 

93 kwonlydefaults: Optional[Dict[str, Any]] 

94 annotations: Mapping[str, Any] 

95 

96 

97def inspect_getfullargspec(func: Callable[..., Any]) -> FullArgSpec: 

98 """Fully vendored version of getfullargspec from Python 3.3.""" 

99 

100 if inspect.ismethod(func): 

101 func = func.__func__ 

102 if not inspect.isfunction(func) and not hasattr(func, "__code__"): 

103 raise TypeError(f"{func!r} is not a Python function") 

104 

105 co = func.__code__ 

106 if not inspect.iscode(co): 

107 raise TypeError(f"{co!r} is not a code object") 

108 

109 nargs = co.co_argcount 

110 names = co.co_varnames 

111 nkwargs = co.co_kwonlyargcount 

112 args = list(names[:nargs]) 

113 kwonlyargs = list(names[nargs : nargs + nkwargs]) 

114 

115 nargs += nkwargs 

116 varargs = None 

117 if co.co_flags & inspect.CO_VARARGS: 

118 varargs = co.co_varnames[nargs] 

119 nargs = nargs + 1 

120 varkw = None 

121 if co.co_flags & inspect.CO_VARKEYWORDS: 

122 varkw = co.co_varnames[nargs] 

123 

124 return FullArgSpec( 

125 args, 

126 varargs, 

127 varkw, 

128 func.__defaults__, 

129 kwonlyargs, 

130 func.__kwdefaults__, 

131 get_annotations(func), 

132 ) 

133 

134 

135# python stubs don't have a public type for this. not worth 

136# making a protocol 

137def md5_not_for_security() -> Any: 

138 return hashlib.md5(usedforsecurity=False) 

139 

140 

141def importlib_metadata_get(group): 

142 ep = importlib_metadata.entry_points() 

143 if typing.TYPE_CHECKING or hasattr(ep, "select"): 

144 return ep.select(group=group) 

145 else: 

146 return ep.get(group, ()) 

147 

148 

149def b(s): 

150 return s.encode("latin-1") 

151 

152 

153def b64decode(x: str) -> bytes: 

154 return base64.b64decode(x.encode("ascii")) 

155 

156 

157def b64encode(x: bytes) -> str: 

158 return base64.b64encode(x).decode("ascii") 

159 

160 

161def decode_backslashreplace(text: bytes, encoding: str) -> str: 

162 return text.decode(encoding, errors="backslashreplace") 

163 

164 

165def cmp(a, b): 

166 return (a > b) - (a < b) 

167 

168 

169def _formatannotation(annotation, base_module=None): 

170 """vendored from python 3.7""" 

171 

172 if isinstance(annotation, str): 

173 return annotation 

174 

175 if getattr(annotation, "__module__", None) == "typing": 

176 return repr(annotation).replace("typing.", "").replace("~", "") 

177 if isinstance(annotation, type): 

178 if annotation.__module__ in ("builtins", base_module): 

179 return repr(annotation.__qualname__) 

180 return annotation.__module__ + "." + annotation.__qualname__ 

181 elif isinstance(annotation, typing.TypeVar): 

182 return repr(annotation).replace("~", "") 

183 return repr(annotation).replace("~", "") 

184 

185 

186def inspect_formatargspec( 

187 args: List[str], 

188 varargs: Optional[str] = None, 

189 varkw: Optional[str] = None, 

190 defaults: Optional[Sequence[Any]] = None, 

191 kwonlyargs: Optional[Sequence[str]] = (), 

192 kwonlydefaults: Optional[Mapping[str, Any]] = {}, 

193 annotations: Mapping[str, Any] = {}, 

194 formatarg: Callable[[str], str] = str, 

195 formatvarargs: Callable[[str], str] = lambda name: "*" + name, 

196 formatvarkw: Callable[[str], str] = lambda name: "**" + name, 

197 formatvalue: Callable[[Any], str] = lambda value: "=" + repr(value), 

198 formatreturns: Callable[[Any], str] = lambda text: " -> " + str(text), 

199 formatannotation: Callable[[Any], str] = _formatannotation, 

200) -> str: 

201 """Copy formatargspec from python 3.7 standard library. 

202 

203 Python 3 has deprecated formatargspec and requested that Signature 

204 be used instead, however this requires a full reimplementation 

205 of formatargspec() in terms of creating Parameter objects and such. 

206 Instead of introducing all the object-creation overhead and having 

207 to reinvent from scratch, just copy their compatibility routine. 

208 

209 Ultimately we would need to rewrite our "decorator" routine completely 

210 which is not really worth it right now, until all Python 2.x support 

211 is dropped. 

212 

213 """ 

214 

215 kwonlydefaults = kwonlydefaults or {} 

216 annotations = annotations or {} 

217 

218 def formatargandannotation(arg): 

219 result = formatarg(arg) 

220 if arg in annotations: 

221 result += ": " + formatannotation(annotations[arg]) 

222 return result 

223 

224 specs = [] 

225 if defaults: 

226 firstdefault = len(args) - len(defaults) 

227 else: 

228 firstdefault = -1 

229 

230 for i, arg in enumerate(args): 

231 spec = formatargandannotation(arg) 

232 if defaults and i >= firstdefault: 

233 spec = spec + formatvalue(defaults[i - firstdefault]) 

234 specs.append(spec) 

235 

236 if varargs is not None: 

237 specs.append(formatvarargs(formatargandannotation(varargs))) 

238 else: 

239 if kwonlyargs: 

240 specs.append("*") 

241 

242 if kwonlyargs: 

243 for kwonlyarg in kwonlyargs: 

244 spec = formatargandannotation(kwonlyarg) 

245 if kwonlydefaults and kwonlyarg in kwonlydefaults: 

246 spec += formatvalue(kwonlydefaults[kwonlyarg]) 

247 specs.append(spec) 

248 

249 if varkw is not None: 

250 specs.append(formatvarkw(formatargandannotation(varkw))) 

251 

252 result = "(" + ", ".join(specs) + ")" 

253 if "return" in annotations: 

254 result += formatreturns(formatannotation(annotations["return"])) 

255 return result 

256 

257 

258def dataclass_fields(cls: Type[Any]) -> Iterable[dataclasses.Field[Any]]: 

259 """Return a sequence of all dataclasses.Field objects associated 

260 with a class as an already processed dataclass. 

261 

262 The class must **already be a dataclass** for Field objects to be returned. 

263 

264 """ 

265 

266 if dataclasses.is_dataclass(cls): 

267 return dataclasses.fields(cls) 

268 else: 

269 return [] 

270 

271 

272def local_dataclass_fields(cls: Type[Any]) -> Iterable[dataclasses.Field[Any]]: 

273 """Return a sequence of all dataclasses.Field objects associated with 

274 an already processed dataclass, excluding those that originate from a 

275 superclass. 

276 

277 The class must **already be a dataclass** for Field objects to be returned. 

278 

279 """ 

280 

281 if dataclasses.is_dataclass(cls): 

282 super_fields: Set[dataclasses.Field[Any]] = set() 

283 for sup in cls.__bases__: 

284 super_fields.update(dataclass_fields(sup)) 

285 return [f for f in dataclasses.fields(cls) if f not in super_fields] 

286 else: 

287 return [] 

288 

289 

290if freethreading: 

291 import threading 

292 

293 mini_gil = threading.RLock() 

294 """provide a threading.RLock() under python freethreading only""" 

295else: 

296 import contextlib 

297 

298 mini_gil = contextlib.nullcontext() # type: ignore[assignment]