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

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 asyncio 

14import base64 

15import dataclasses 

16import hashlib 

17from importlib import metadata as importlib_metadata 

18import inspect 

19import operator 

20import platform 

21import sys 

22import sysconfig 

23import typing 

24from typing import Any 

25from typing import Awaitable 

26from typing import Callable 

27from typing import Dict 

28from typing import Iterable 

29from typing import List 

30from typing import Mapping 

31from typing import Optional 

32from typing import Sequence 

33from typing import Set 

34from typing import Tuple 

35from typing import Type 

36from typing import TypeVar 

37 

38_T = TypeVar("_T") 

39 

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

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

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

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

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

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

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

47 

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

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

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

51is64bit = sys.maxsize > 2**32 

52 

53has_refcount_gc = bool(cpython) 

54 

55dottedgetter = operator.attrgetter 

56 

57 

58# use sys.version_info to enable mypy version narrowing 

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

60 

61 import annotationlib 

62 from string.templatelib import Template as Template 

63 

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

65 return annotationlib.get_annotations( 

66 obj, format=annotationlib.Format.FORWARDREF 

67 ) 

68 

69else: 

70 

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

72 return inspect.get_annotations(obj) 

73 

74 class Template: 

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

76 

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

78 self._parts = parts 

79 

80 @property 

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

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

83 

84 @property 

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

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

87 

88 def __iter__(self) -> Any: 

89 return iter(self._parts) 

90 

91 

92class FullArgSpec(typing.NamedTuple): 

93 args: List[str] 

94 varargs: Optional[str] 

95 varkw: Optional[str] 

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

97 kwonlyargs: List[str] 

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

99 annotations: Mapping[str, Any] 

100 

101 

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

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

104 

105 if inspect.ismethod(func): 

106 func = func.__func__ 

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

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

109 

110 co = func.__code__ 

111 if not inspect.iscode(co): 

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

113 

114 nargs = co.co_argcount 

115 names = co.co_varnames 

116 nkwargs = co.co_kwonlyargcount 

117 args = list(names[:nargs]) 

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

119 

120 nargs += nkwargs 

121 varargs = None 

122 if co.co_flags & inspect.CO_VARARGS: 

123 varargs = co.co_varnames[nargs] 

124 nargs = nargs + 1 

125 varkw = None 

126 if co.co_flags & inspect.CO_VARKEYWORDS: 

127 varkw = co.co_varnames[nargs] 

128 

129 return FullArgSpec( 

130 args, 

131 varargs, 

132 varkw, 

133 func.__defaults__, 

134 kwonlyargs, 

135 func.__kwdefaults__, 

136 get_annotations(func), 

137 ) 

138 

139 

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

141# making a protocol 

142def md5_not_for_security() -> Any: 

143 return hashlib.md5(usedforsecurity=False) 

144 

145 

146def importlib_metadata_get(group): 

147 ep = importlib_metadata.entry_points() 

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

149 return ep.select(group=group) 

150 else: 

151 return ep.get(group, ()) 

152 

153 

154def b(s): 

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

156 

157 

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

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

160 

161 

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

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

164 

165 

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

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

168 

169 

170def cmp(a, b): 

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

172 

173 

174def _formatannotation(annotation, base_module=None): 

175 """vendored from python 3.7""" 

176 

177 if isinstance(annotation, str): 

178 return annotation 

179 

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

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

182 if isinstance(annotation, type): 

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

184 return repr(annotation.__qualname__) 

185 return annotation.__module__ + "." + annotation.__qualname__ 

186 elif isinstance(annotation, typing.TypeVar): 

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

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

189 

190 

191def inspect_formatargspec( 

192 args: List[str], 

193 varargs: Optional[str] = None, 

194 varkw: Optional[str] = None, 

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

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

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

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

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

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

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

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

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

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

205) -> str: 

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

207 

208 Python 3 has deprecated formatargspec and requested that Signature 

209 be used instead, however this requires a full reimplementation 

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

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

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

213 

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

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

216 is dropped. 

217 

218 """ 

219 

220 kwonlydefaults = kwonlydefaults or {} 

221 annotations = annotations or {} 

222 

223 def formatargandannotation(arg): 

224 result = formatarg(arg) 

225 if arg in annotations: 

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

227 return result 

228 

229 specs = [] 

230 if defaults: 

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

232 else: 

233 firstdefault = -1 

234 

235 for i, arg in enumerate(args): 

236 spec = formatargandannotation(arg) 

237 if defaults and i >= firstdefault: 

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

239 specs.append(spec) 

240 

241 if varargs is not None: 

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

243 else: 

244 if kwonlyargs: 

245 specs.append("*") 

246 

247 if kwonlyargs: 

248 for kwonlyarg in kwonlyargs: 

249 spec = formatargandannotation(kwonlyarg) 

250 if kwonlydefaults and kwonlyarg in kwonlydefaults: 

251 spec += formatvalue(kwonlydefaults[kwonlyarg]) 

252 specs.append(spec) 

253 

254 if varkw is not None: 

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

256 

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

258 if "return" in annotations: 

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

260 return result 

261 

262 

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

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

265 with a class as an already processed dataclass. 

266 

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

268 

269 """ 

270 

271 if dataclasses.is_dataclass(cls): 

272 return dataclasses.fields(cls) 

273 else: 

274 return [] 

275 

276 

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

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

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

280 superclass. 

281 

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

283 

284 """ 

285 

286 if dataclasses.is_dataclass(cls): 

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

288 for sup in cls.__bases__: 

289 super_fields.update(dataclass_fields(sup)) 

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

291 else: 

292 return [] 

293 

294 

295if freethreading: 

296 import threading 

297 

298 mini_gil = threading.RLock() 

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

300else: 

301 import contextlib 

302 

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

304 

305 

306# use sys.version_info to enable mypy version narrowing 

307if sys.version_info >= (3, 12): 

308 

309 async def wait_for( 

310 timeout: float, 

311 fn: Callable[..., Awaitable[_T]], 

312 *arg: Any, 

313 **kw: Any, 

314 ) -> _T: 

315 """Await ``fn(*arg, **kw)`` with a timeout, raising 

316 ``TimeoutError`` when the timeout is reached. 

317 

318 """ 

319 return await asyncio.wait_for(fn(*arg, **kw), timeout) 

320 

321else: 

322 

323 async def wait_for( 

324 timeout: float, 

325 fn: Callable[..., Awaitable[_T]], 

326 *arg: Any, 

327 **kw: Any, 

328 ) -> _T: 

329 # on Python 3.11, asyncio.wait_for() returns normally in place of 

330 # raising a cancellation that reaches it once the inner operation 

331 # has completed; cpython issue gh-86296, fixed in Python 3.12. 

332 # this version can be removed when Python 3.11 support is dropped 

333 async with asyncio.timeout(timeout): 

334 return await fn(*arg, **kw)