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 

34from typing import TYPE_CHECKING 

35 

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

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

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

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

40py311 = sys.version_info >= (3, 11) 

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

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

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

44 

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

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

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

48is64bit = sys.maxsize > 2**32 

49 

50has_refcount_gc = bool(cpython) 

51 

52dottedgetter = operator.attrgetter 

53 

54if py314 or TYPE_CHECKING: 

55 from string.templatelib import Template as Template 

56else: 

57 

58 class Template: # type: ignore[no-redef] 

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

60 

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

62 self._parts = parts 

63 

64 @property 

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

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

67 

68 @property 

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

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

71 

72 def __iter__(self) -> Any: 

73 return iter(self._parts) 

74 

75 

76if py314: 

77 

78 import annotationlib 

79 

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

81 return annotationlib.get_annotations( 

82 obj, format=annotationlib.Format.FORWARDREF 

83 ) 

84 

85else: 

86 

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

88 return inspect.get_annotations(obj) 

89 

90 

91class FullArgSpec(typing.NamedTuple): 

92 args: List[str] 

93 varargs: Optional[str] 

94 varkw: Optional[str] 

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

96 kwonlyargs: List[str] 

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

98 annotations: Mapping[str, Any] 

99 

100 

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

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

103 

104 if inspect.ismethod(func): 

105 func = func.__func__ 

106 if not inspect.isfunction(func): 

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

108 

109 co = func.__code__ 

110 if not inspect.iscode(co): 

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

112 

113 nargs = co.co_argcount 

114 names = co.co_varnames 

115 nkwargs = co.co_kwonlyargcount 

116 args = list(names[:nargs]) 

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

118 

119 nargs += nkwargs 

120 varargs = None 

121 if co.co_flags & inspect.CO_VARARGS: 

122 varargs = co.co_varnames[nargs] 

123 nargs = nargs + 1 

124 varkw = None 

125 if co.co_flags & inspect.CO_VARKEYWORDS: 

126 varkw = co.co_varnames[nargs] 

127 

128 return FullArgSpec( 

129 args, 

130 varargs, 

131 varkw, 

132 func.__defaults__, 

133 kwonlyargs, 

134 func.__kwdefaults__, 

135 get_annotations(func), 

136 ) 

137 

138 

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

140# making a protocol 

141def md5_not_for_security() -> Any: 

142 return hashlib.md5(usedforsecurity=False) 

143 

144 

145def importlib_metadata_get(group): 

146 ep = importlib_metadata.entry_points() 

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

148 return ep.select(group=group) 

149 else: 

150 return ep.get(group, ()) 

151 

152 

153def b(s): 

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

155 

156 

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

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

159 

160 

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

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

163 

164 

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

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

167 

168 

169def cmp(a, b): 

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

171 

172 

173def _formatannotation(annotation, base_module=None): 

174 """vendored from python 3.7""" 

175 

176 if isinstance(annotation, str): 

177 return annotation 

178 

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

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

181 if isinstance(annotation, type): 

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

183 return repr(annotation.__qualname__) 

184 return annotation.__module__ + "." + annotation.__qualname__ 

185 elif isinstance(annotation, typing.TypeVar): 

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

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

188 

189 

190def inspect_formatargspec( 

191 args: List[str], 

192 varargs: Optional[str] = None, 

193 varkw: Optional[str] = None, 

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

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

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

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

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

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

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

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

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

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

204) -> str: 

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

206 

207 Python 3 has deprecated formatargspec and requested that Signature 

208 be used instead, however this requires a full reimplementation 

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

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

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

212 

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

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

215 is dropped. 

216 

217 """ 

218 

219 kwonlydefaults = kwonlydefaults or {} 

220 annotations = annotations or {} 

221 

222 def formatargandannotation(arg): 

223 result = formatarg(arg) 

224 if arg in annotations: 

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

226 return result 

227 

228 specs = [] 

229 if defaults: 

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

231 else: 

232 firstdefault = -1 

233 

234 for i, arg in enumerate(args): 

235 spec = formatargandannotation(arg) 

236 if defaults and i >= firstdefault: 

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

238 specs.append(spec) 

239 

240 if varargs is not None: 

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

242 else: 

243 if kwonlyargs: 

244 specs.append("*") 

245 

246 if kwonlyargs: 

247 for kwonlyarg in kwonlyargs: 

248 spec = formatargandannotation(kwonlyarg) 

249 if kwonlydefaults and kwonlyarg in kwonlydefaults: 

250 spec += formatvalue(kwonlydefaults[kwonlyarg]) 

251 specs.append(spec) 

252 

253 if varkw is not None: 

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

255 

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

257 if "return" in annotations: 

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

259 return result 

260 

261 

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

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

264 with a class as an already processed dataclass. 

265 

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

267 

268 """ 

269 

270 if dataclasses.is_dataclass(cls): 

271 return dataclasses.fields(cls) 

272 else: 

273 return [] 

274 

275 

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

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

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

279 superclass. 

280 

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

282 

283 """ 

284 

285 if dataclasses.is_dataclass(cls): 

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

287 for sup in cls.__bases__: 

288 super_fields.update(dataclass_fields(sup)) 

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

290 else: 

291 return [] 

292 

293 

294if freethreading: 

295 import threading 

296 

297 mini_gil = threading.RLock() 

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

299else: 

300 import contextlib 

301 

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