codekingpro/portable-devtools

1# util/compat.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# 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
16import inspect
17import operator
18import platform
19import sys
20import typing
21from typing import Any
22from typing import Callable
23from typing import Dict
24from typing import Iterable
25from typing import List
26from typing import Mapping
27from typing import Optional
28from typing import Sequence
29from typing import Set
30from typing import Tuple
31from typing import Type
32from typing import TypeVar
33
34
35py312 = sys.version_info >= (3, 12)
36py311 = sys.version_info >= (3, 11)
37py310 = sys.version_info >= (3, 10)
38py39 = sys.version_info >= (3, 9)
39py38 = sys.version_info >= (3, 8)
40pypy = platform.python_implementation() == "PyPy"
41cpython = platform.python_implementation() == "CPython"
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_T_co = TypeVar("_T_co", covariant=True)
53
54
55class FullArgSpec(typing.NamedTuple):
56    args: List[str]
57    varargs: Optional[str]
58    varkw: Optional[str]
59    defaults: Optional[Tuple[Any, ...]]
60    kwonlyargs: List[str]
61    kwonlydefaults: Dict[str, Any]
62    annotations: Dict[str, Any]
63
64
65def inspect_getfullargspec(func: Callable[..., Any]) -> FullArgSpec:
66    """Fully vendored version of getfullargspec from Python 3.3."""
67
68    if inspect.ismethod(func):
69        func = func.__func__
70    if not inspect.isfunction(func):
71        raise TypeError(f"{func!r} is not a Python function")
72
73    co = func.__code__
74    if not inspect.iscode(co):
75        raise TypeError(f"{co!r} is not a code object")
76
77    nargs = co.co_argcount
78    names = co.co_varnames
79    nkwargs = co.co_kwonlyargcount
80    args = list(names[:nargs])
81    kwonlyargs = list(names[nargs : nargs + nkwargs])
82
83    nargs += nkwargs
84    varargs = None
85    if co.co_flags & inspect.CO_VARARGS:
86        varargs = co.co_varnames[nargs]
87        nargs = nargs + 1
88    varkw = None
89    if co.co_flags & inspect.CO_VARKEYWORDS:
90        varkw = co.co_varnames[nargs]
91
92    return FullArgSpec(
93        args,
94        varargs,
95        varkw,
96        func.__defaults__,
97        kwonlyargs,
98        func.__kwdefaults__,
99        func.__annotations__,
100    )
101
102
103if py39:
104    # python stubs don't have a public type for this. not worth
105    # making a protocol
106    def md5_not_for_security() -> Any:
107        return hashlib.md5(usedforsecurity=False)
108
109else:
110
111    def md5_not_for_security() -> Any:
112        return hashlib.md5()
113
114
115if typing.TYPE_CHECKING or py38:
116    from importlib import metadata as importlib_metadata
117else:
118    import importlib_metadata  # noqa
119
120
121if typing.TYPE_CHECKING or py39:
122    # pep 584 dict union
123    dict_union = operator.or_  # noqa
124else:
125
126    def dict_union(a: dict, b: dict) -> dict:
127        a = a.copy()
128        a.update(b)
129        return a
130
131
132if py310:
133    anext_ = anext
134else:
135    _NOT_PROVIDED = object()
136    from collections.abc import AsyncIterator
137
138    async def anext_(async_iterator, default=_NOT_PROVIDED):
139        """vendored from https://github.com/python/cpython/pull/8895"""
140
141        if not isinstance(async_iterator, AsyncIterator):
142            raise TypeError(
143                f"anext expected an AsyncIterator, got {type(async_iterator)}"
144            )
145        anxt = type(async_iterator).__anext__
146        try:
147            return await anxt(async_iterator)
148        except StopAsyncIteration:
149            if default is _NOT_PROVIDED:
150                raise
151            return default
152
153
154def importlib_metadata_get(group):
155    ep = importlib_metadata.entry_points()
156    if typing.TYPE_CHECKING or hasattr(ep, "select"):
157        return ep.select(group=group)
158    else:
159        return ep.get(group, ())
160
161
162def b(s):
163    return s.encode("latin-1")
164
165
166def b64decode(x: str) -> bytes:
167    return base64.b64decode(x.encode("ascii"))
168
169
170def b64encode(x: bytes) -> str:
171    return base64.b64encode(x).decode("ascii")
172
173
174def decode_backslashreplace(text: bytes, encoding: str) -> str:
175    return text.decode(encoding, errors="backslashreplace")
176
177
178def cmp(a, b):
179    return (a > b) - (a < b)
180
181
182def _formatannotation(annotation, base_module=None):
183    """vendored from python 3.7"""
184
185    if isinstance(annotation, str):
186        return annotation
187
188    if getattr(annotation, "__module__", None) == "typing":
189        return repr(annotation).replace("typing.", "").replace("~", "")
190    if isinstance(annotation, type):
191        if annotation.__module__ in ("builtins", base_module):
192            return repr(annotation.__qualname__)
193        return annotation.__module__ + "." + annotation.__qualname__
194    elif isinstance(annotation, typing.TypeVar):
195        return repr(annotation).replace("~", "")
196    return repr(annotation).replace("~", "")
197
198
199def inspect_formatargspec(
200    args: List[str],
201    varargs: Optional[str] = None,
202    varkw: Optional[str] = None,
203    defaults: Optional[Sequence[Any]] = None,
204    kwonlyargs: Optional[Sequence[str]] = (),
205    kwonlydefaults: Optional[Mapping[str, Any]] = {},
206    annotations: Mapping[str, Any] = {},
207    formatarg: Callable[[str], str] = str,
208    formatvarargs: Callable[[str], str] = lambda name: "*" + name,
209    formatvarkw: Callable[[str], str] = lambda name: "**" + name,
210    formatvalue: Callable[[Any], str] = lambda value: "=" + repr(value),
211    formatreturns: Callable[[Any], str] = lambda text: " -> " + str(text),
212    formatannotation: Callable[[Any], str] = _formatannotation,
213) -> str:
214    """Copy formatargspec from python 3.7 standard library.
215
216    Python 3 has deprecated formatargspec and requested that Signature
217    be used instead, however this requires a full reimplementation
218    of formatargspec() in terms of creating Parameter objects and such.
219    Instead of introducing all the object-creation overhead and having
220    to reinvent from scratch, just copy their compatibility routine.
221
222    Ultimately we would need to rewrite our "decorator" routine completely
223    which is not really worth it right now, until all Python 2.x support
224    is dropped.
225
226    """
227
228    kwonlydefaults = kwonlydefaults or {}
229    annotations = annotations or {}
230
231    def formatargandannotation(arg):
232        result = formatarg(arg)
233        if arg in annotations:
234            result += ": " + formatannotation(annotations[arg])
235        return result
236
237    specs = []
238    if defaults:
239        firstdefault = len(args) - len(defaults)
240    else:
241        firstdefault = -1
242
243    for i, arg in enumerate(args):
244        spec = formatargandannotation(arg)
245        if defaults and i >= firstdefault:
246            spec = spec + formatvalue(defaults[i - firstdefault])
247        specs.append(spec)
248
249    if varargs is not None:
250        specs.append(formatvarargs(formatargandannotation(varargs)))
251    else:
252        if kwonlyargs:
253            specs.append("*")
254
255    if kwonlyargs:
256        for kwonlyarg in kwonlyargs:
257            spec = formatargandannotation(kwonlyarg)
258            if kwonlydefaults and kwonlyarg in kwonlydefaults:
259                spec += formatvalue(kwonlydefaults[kwonlyarg])
260            specs.append(spec)
261
262    if varkw is not None:
263        specs.append(formatvarkw(formatargandannotation(varkw)))
264
265    result = "(" + ", ".join(specs) + ")"
266    if "return" in annotations:
267        result += formatreturns(formatannotation(annotations["return"]))
268    return result
269
270
271def dataclass_fields(cls: Type[Any]) -> Iterable[dataclasses.Field[Any]]:
272    """Return a sequence of all dataclasses.Field objects associated
273    with a class as an already processed dataclass.
274
275    The class must **already be a dataclass** for Field objects to be returned.
276
277    """
278
279    if dataclasses.is_dataclass(cls):
280        return dataclasses.fields(cls)
281    else:
282        return []
283
284
285def local_dataclass_fields(cls: Type[Any]) -> Iterable[dataclasses.Field[Any]]:
286    """Return a sequence of all dataclasses.Field objects associated with
287    an already processed dataclass, excluding those that originate from a
288    superclass.
289
290    The class must **already be a dataclass** for Field objects to be returned.
291
292    """
293
294    if dataclasses.is_dataclass(cls):
295        super_fields: Set[dataclasses.Field[Any]] = set()
296        for sup in cls.__bases__:
297            super_fields.update(dataclass_fields(sup))
298        return [f for f in dataclasses.fields(cls) if f not in super_fields]
299    else:
300        return []
301