compat.py 13 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421
  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. """Handle Python version/platform incompatibilities."""
  9. from __future__ import annotations
  10. import base64
  11. import dataclasses
  12. import hashlib
  13. import inspect
  14. import operator
  15. import platform
  16. import sys
  17. import sysconfig
  18. import typing
  19. from typing import Any
  20. from typing import Callable
  21. from typing import cast
  22. from typing import Dict
  23. from typing import Iterable
  24. from typing import List
  25. from typing import Mapping
  26. from typing import Optional
  27. from typing import Sequence
  28. from typing import Set
  29. from typing import Tuple
  30. from typing import Type
  31. from typing import TypeVar
  32. py314b1 = sys.version_info >= (3, 14, 0, "beta", 1)
  33. py314 = sys.version_info >= (3, 14)
  34. py313 = sys.version_info >= (3, 13)
  35. py312 = sys.version_info >= (3, 12)
  36. py311 = sys.version_info >= (3, 11)
  37. py310 = sys.version_info >= (3, 10)
  38. py39 = sys.version_info >= (3, 9)
  39. py38 = sys.version_info >= (3, 8)
  40. pypy = platform.python_implementation() == "PyPy"
  41. cpython = platform.python_implementation() == "CPython"
  42. freethreading = bool(sysconfig.get_config_var("Py_GIL_DISABLED"))
  43. win32 = sys.platform.startswith("win")
  44. osx = sys.platform.startswith("darwin")
  45. arm = "aarch" in platform.machine().lower()
  46. is64bit = sys.maxsize > 2**32
  47. has_refcount_gc = bool(cpython)
  48. dottedgetter = operator.attrgetter
  49. _T_co = TypeVar("_T_co", covariant=True)
  50. if py314:
  51. # vendor a minimal form of get_annotations per
  52. # https://github.com/python/cpython/issues/133684#issuecomment-2863841891
  53. from annotationlib import call_annotate_function # type: ignore[import-not-found,unused-ignore] # noqa: E501
  54. from annotationlib import Format
  55. def _get_and_call_annotate(obj, format): # noqa: A002
  56. annotate = getattr(obj, "__annotate__", None)
  57. if annotate is not None:
  58. ann = call_annotate_function(annotate, format, owner=obj)
  59. if not isinstance(ann, dict):
  60. raise ValueError(f"{obj!r}.__annotate__ returned a non-dict")
  61. return ann
  62. return None
  63. # this is ported from py3.13.0a7
  64. _BASE_GET_ANNOTATIONS = type.__dict__["__annotations__"].__get__
  65. def _get_dunder_annotations(obj):
  66. if isinstance(obj, type):
  67. try:
  68. ann = _BASE_GET_ANNOTATIONS(obj)
  69. except AttributeError:
  70. # For static types, the descriptor raises AttributeError.
  71. return {}
  72. else:
  73. ann = getattr(obj, "__annotations__", None)
  74. if ann is None:
  75. return {}
  76. if not isinstance(ann, dict):
  77. raise ValueError(
  78. f"{obj!r}.__annotations__ is neither a dict nor None"
  79. )
  80. return dict(ann)
  81. def _vendored_get_annotations(
  82. obj: Any, *, format: Format # noqa: A002
  83. ) -> Mapping[str, Any]:
  84. """A sparse implementation of annotationlib.get_annotations()"""
  85. try:
  86. ann = _get_dunder_annotations(obj)
  87. except Exception:
  88. pass
  89. else:
  90. if ann is not None:
  91. return dict(ann)
  92. # But if __annotations__ threw a NameError, we try calling __annotate__
  93. ann = _get_and_call_annotate(obj, format)
  94. if ann is None:
  95. # If that didn't work either, we have a very weird object:
  96. # evaluating
  97. # __annotations__ threw NameError and there is no __annotate__.
  98. # In that case,
  99. # we fall back to trying __annotations__ again.
  100. ann = _get_dunder_annotations(obj)
  101. if ann is None:
  102. if isinstance(obj, type) or callable(obj):
  103. return {}
  104. raise TypeError(f"{obj!r} does not have annotations")
  105. if not ann:
  106. return {}
  107. return dict(ann)
  108. def get_annotations(obj: Any) -> Mapping[str, Any]:
  109. # FORWARDREF has the effect of giving us ForwardRefs and not
  110. # actually trying to evaluate the annotations. We need this so
  111. # that the annotations act as much like
  112. # "from __future__ import annotations" as possible, which is going
  113. # away in future python as a separate mode
  114. return _vendored_get_annotations(obj, format=Format.FORWARDREF)
  115. elif py310:
  116. def get_annotations(obj: Any) -> Mapping[str, Any]:
  117. return inspect.get_annotations(obj)
  118. else:
  119. def get_annotations(obj: Any) -> Mapping[str, Any]:
  120. # it's been observed that cls.__annotations__ can be non present.
  121. # it's not clear what causes this, running under tox py37/38 it
  122. # happens, running straight pytest it doesnt
  123. # https://docs.python.org/3/howto/annotations.html#annotations-howto
  124. if isinstance(obj, type):
  125. ann = obj.__dict__.get("__annotations__", None)
  126. else:
  127. ann = getattr(obj, "__annotations__", None)
  128. from . import _collections
  129. if ann is None:
  130. return _collections.EMPTY_DICT
  131. else:
  132. return cast("Mapping[str, Any]", ann)
  133. class FullArgSpec(typing.NamedTuple):
  134. args: List[str]
  135. varargs: Optional[str]
  136. varkw: Optional[str]
  137. defaults: Optional[Tuple[Any, ...]]
  138. kwonlyargs: List[str]
  139. kwonlydefaults: Optional[Dict[str, Any]]
  140. annotations: Mapping[str, Any]
  141. def inspect_getfullargspec(func: Callable[..., Any]) -> FullArgSpec:
  142. """Fully vendored version of getfullargspec from Python 3.3."""
  143. if inspect.ismethod(func):
  144. func = func.__func__
  145. if not inspect.isfunction(func):
  146. raise TypeError(f"{func!r} is not a Python function")
  147. co = func.__code__
  148. if not inspect.iscode(co):
  149. raise TypeError(f"{co!r} is not a code object")
  150. nargs = co.co_argcount
  151. names = co.co_varnames
  152. nkwargs = co.co_kwonlyargcount
  153. args = list(names[:nargs])
  154. kwonlyargs = list(names[nargs : nargs + nkwargs])
  155. nargs += nkwargs
  156. varargs = None
  157. if co.co_flags & inspect.CO_VARARGS:
  158. varargs = co.co_varnames[nargs]
  159. nargs = nargs + 1
  160. varkw = None
  161. if co.co_flags & inspect.CO_VARKEYWORDS:
  162. varkw = co.co_varnames[nargs]
  163. return FullArgSpec(
  164. args,
  165. varargs,
  166. varkw,
  167. func.__defaults__,
  168. kwonlyargs,
  169. func.__kwdefaults__,
  170. get_annotations(func),
  171. )
  172. if py39:
  173. # python stubs don't have a public type for this. not worth
  174. # making a protocol
  175. def md5_not_for_security() -> Any:
  176. return hashlib.md5(usedforsecurity=False)
  177. else:
  178. def md5_not_for_security() -> Any:
  179. return hashlib.md5()
  180. if typing.TYPE_CHECKING or py38:
  181. from importlib import metadata as importlib_metadata
  182. else:
  183. import importlib_metadata # noqa
  184. if typing.TYPE_CHECKING or py39:
  185. # pep 584 dict union
  186. dict_union = operator.or_ # noqa
  187. else:
  188. def dict_union(a: dict, b: dict) -> dict:
  189. a = a.copy()
  190. a.update(b)
  191. return a
  192. if py310:
  193. anext_ = anext
  194. else:
  195. _NOT_PROVIDED = object()
  196. from collections.abc import AsyncIterator
  197. async def anext_(async_iterator, default=_NOT_PROVIDED):
  198. """vendored from https://github.com/python/cpython/pull/8895"""
  199. if not isinstance(async_iterator, AsyncIterator):
  200. raise TypeError(
  201. f"anext expected an AsyncIterator, got {type(async_iterator)}"
  202. )
  203. anxt = type(async_iterator).__anext__
  204. try:
  205. return await anxt(async_iterator)
  206. except StopAsyncIteration:
  207. if default is _NOT_PROVIDED:
  208. raise
  209. return default
  210. def importlib_metadata_get(group):
  211. ep = importlib_metadata.entry_points()
  212. if typing.TYPE_CHECKING or hasattr(ep, "select"):
  213. return ep.select(group=group)
  214. else:
  215. return ep.get(group, ())
  216. def b(s):
  217. return s.encode("latin-1")
  218. def b64decode(x: str) -> bytes:
  219. return base64.b64decode(x.encode("ascii"))
  220. def b64encode(x: bytes) -> str:
  221. return base64.b64encode(x).decode("ascii")
  222. def decode_backslashreplace(text: bytes, encoding: str) -> str:
  223. return text.decode(encoding, errors="backslashreplace")
  224. def cmp(a, b):
  225. return (a > b) - (a < b)
  226. def _formatannotation(annotation, base_module=None):
  227. """vendored from python 3.7"""
  228. if isinstance(annotation, str):
  229. return annotation
  230. if getattr(annotation, "__module__", None) == "typing":
  231. return repr(annotation).replace("typing.", "").replace("~", "")
  232. if isinstance(annotation, type):
  233. if annotation.__module__ in ("builtins", base_module):
  234. return repr(annotation.__qualname__)
  235. return annotation.__module__ + "." + annotation.__qualname__
  236. elif isinstance(annotation, typing.TypeVar):
  237. return repr(annotation).replace("~", "")
  238. return repr(annotation).replace("~", "")
  239. def inspect_formatargspec(
  240. args: List[str],
  241. varargs: Optional[str] = None,
  242. varkw: Optional[str] = None,
  243. defaults: Optional[Sequence[Any]] = None,
  244. kwonlyargs: Optional[Sequence[str]] = (),
  245. kwonlydefaults: Optional[Mapping[str, Any]] = {},
  246. annotations: Mapping[str, Any] = {},
  247. formatarg: Callable[[str], str] = str,
  248. formatvarargs: Callable[[str], str] = lambda name: "*" + name,
  249. formatvarkw: Callable[[str], str] = lambda name: "**" + name,
  250. formatvalue: Callable[[Any], str] = lambda value: "=" + repr(value),
  251. formatreturns: Callable[[Any], str] = lambda text: " -> " + str(text),
  252. formatannotation: Callable[[Any], str] = _formatannotation,
  253. ) -> str:
  254. """Copy formatargspec from python 3.7 standard library.
  255. Python 3 has deprecated formatargspec and requested that Signature
  256. be used instead, however this requires a full reimplementation
  257. of formatargspec() in terms of creating Parameter objects and such.
  258. Instead of introducing all the object-creation overhead and having
  259. to reinvent from scratch, just copy their compatibility routine.
  260. Ultimately we would need to rewrite our "decorator" routine completely
  261. which is not really worth it right now, until all Python 2.x support
  262. is dropped.
  263. """
  264. kwonlydefaults = kwonlydefaults or {}
  265. annotations = annotations or {}
  266. def formatargandannotation(arg):
  267. result = formatarg(arg)
  268. if arg in annotations:
  269. result += ": " + formatannotation(annotations[arg])
  270. return result
  271. specs = []
  272. if defaults:
  273. firstdefault = len(args) - len(defaults)
  274. else:
  275. firstdefault = -1
  276. for i, arg in enumerate(args):
  277. spec = formatargandannotation(arg)
  278. if defaults and i >= firstdefault:
  279. spec = spec + formatvalue(defaults[i - firstdefault])
  280. specs.append(spec)
  281. if varargs is not None:
  282. specs.append(formatvarargs(formatargandannotation(varargs)))
  283. else:
  284. if kwonlyargs:
  285. specs.append("*")
  286. if kwonlyargs:
  287. for kwonlyarg in kwonlyargs:
  288. spec = formatargandannotation(kwonlyarg)
  289. if kwonlydefaults and kwonlyarg in kwonlydefaults:
  290. spec += formatvalue(kwonlydefaults[kwonlyarg])
  291. specs.append(spec)
  292. if varkw is not None:
  293. specs.append(formatvarkw(formatargandannotation(varkw)))
  294. result = "(" + ", ".join(specs) + ")"
  295. if "return" in annotations:
  296. result += formatreturns(formatannotation(annotations["return"]))
  297. return result
  298. def dataclass_fields(cls: Type[Any]) -> Iterable[dataclasses.Field[Any]]:
  299. """Return a sequence of all dataclasses.Field objects associated
  300. with a class as an already processed dataclass.
  301. The class must **already be a dataclass** for Field objects to be returned.
  302. """
  303. if dataclasses.is_dataclass(cls):
  304. return dataclasses.fields(cls)
  305. else:
  306. return []
  307. def local_dataclass_fields(cls: Type[Any]) -> Iterable[dataclasses.Field[Any]]:
  308. """Return a sequence of all dataclasses.Field objects associated with
  309. an already processed dataclass, excluding those that originate from a
  310. superclass.
  311. The class must **already be a dataclass** for Field objects to be returned.
  312. """
  313. if dataclasses.is_dataclass(cls):
  314. super_fields: Set[dataclasses.Field[Any]] = set()
  315. for sup in cls.__bases__:
  316. super_fields.update(dataclass_fields(sup))
  317. return [f for f in dataclasses.fields(cls) if f not in super_fields]
  318. else:
  319. return []
  320. if freethreading:
  321. import threading
  322. mini_gil = threading.RLock()
  323. """provide a threading.RLock() under python freethreading only"""
  324. else:
  325. import contextlib
  326. mini_gil = contextlib.nullcontext() # type: ignore[assignment]