Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1# util/langhelpers.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
9"""Routines to help with the creation, loading and introspection of
10modules, classes, hierarchies, attributes, functions, and methods.
12"""
14from __future__ import annotations
16import collections
17import enum
18from functools import update_wrapper
19import importlib.metadata
20import importlib.util
21import inspect
22import itertools
23import linecache
24import operator
25import re
26import sys
27import textwrap
28import threading
29import types
30from types import CodeType
31from types import ModuleType
32from typing import Any
33from typing import Callable
34from typing import cast
35from typing import Dict
36from typing import FrozenSet
37from typing import Generic
38from typing import Iterator
39from typing import List
40from typing import Literal
41from typing import NoReturn
42from typing import Optional
43from typing import overload
44from typing import Sequence
45from typing import Set
46from typing import Tuple
47from typing import Type
48from typing import TYPE_CHECKING
49from typing import TypeVar
50from typing import Union
51import warnings
52import weakref
54from . import _collections
55from . import compat
56from .. import exc
58_T = TypeVar("_T")
59_T_co = TypeVar("_T_co", covariant=True)
60_F = TypeVar("_F", bound=Callable[..., Any])
61_MA = TypeVar("_MA", bound="HasMemoized.memoized_attribute[Any]")
62_M = TypeVar("_M", bound=ModuleType)
65def restore_annotations(
66 cls: type, new_annotations: dict[str, Any]
67) -> Callable[[], None]:
68 """apply alternate annotations to a class, with a callable to restore
69 the pristine state of the former.
70 This is used strictly to provide dataclasses on a mapped class, where
71 in some cases where are making dataclass fields based on an attribute
72 that is actually a python descriptor on a superclass which we called
73 to get a value.
74 if dataclasses were to give us a way to achieve this without swapping
75 __annotations__, that would be much better.
76 """
77 delattr_ = object()
79 # pep-649 means classes have "__annotate__", and it's a callable. if it's
80 # there and is None, we're in "legacy future mode", where it's python 3.14
81 # or higher and "from __future__ import annotations" is set. in "legacy
82 # future mode" we have to do the same steps we do for older pythons,
83 # __annotate__ can be ignored
84 is_pep649 = hasattr(cls, "__annotate__") and cls.__annotate__ is not None
86 if is_pep649:
87 memoized = {
88 "__annotate__": getattr(cls, "__annotate__", delattr_),
89 }
90 else:
91 memoized = {
92 "__annotations__": getattr(cls, "__annotations__", delattr_)
93 }
95 cls.__annotations__ = new_annotations
97 def restore():
98 for k, v in memoized.items():
99 if v is delattr_:
100 delattr(cls, k)
101 else:
102 setattr(cls, k, v)
104 return restore
107def md5_hex(x: Any) -> str:
108 x = x.encode("utf-8")
109 m = compat.md5_not_for_security()
110 m.update(x)
111 return cast(str, m.hexdigest())
114class safe_reraise:
115 """Reraise an exception after invoking some
116 handler code.
118 Stores the existing exception info before
119 invoking so that it is maintained across a potential
120 coroutine context switch.
122 e.g.::
124 try:
125 sess.commit()
126 except:
127 with safe_reraise():
128 sess.rollback()
130 TODO: we should at some point evaluate current behaviors in this regard
131 based on current greenlet, gevent/eventlet implementations in Python 3, and
132 also see the degree to which our own asyncio (based on greenlet also) is
133 impacted by this. .rollback() will cause IO / context switch to occur in
134 all these scenarios; what happens to the exception context from an
135 "except:" block if we don't explicitly store it? Original issue was #2703.
137 """
139 __slots__ = ("_exc_info",)
141 _exc_info: Union[
142 None,
143 Tuple[
144 Type[BaseException],
145 BaseException,
146 types.TracebackType,
147 ],
148 Tuple[None, None, None],
149 ]
151 def __enter__(self) -> None:
152 self._exc_info = sys.exc_info()
154 def __exit__(
155 self,
156 type_: Optional[Type[BaseException]],
157 value: Optional[BaseException],
158 traceback: Optional[types.TracebackType],
159 ) -> NoReturn:
160 assert self._exc_info is not None
161 # see #2703 for notes
162 if type_ is None:
163 exc_type, exc_value, exc_tb = self._exc_info
164 assert exc_value is not None
165 self._exc_info = None # remove potential circular references
166 raise exc_value.with_traceback(exc_tb)
167 else:
168 self._exc_info = None # remove potential circular references
169 assert value is not None
170 raise value.with_traceback(traceback)
173def walk_subclasses(cls: Type[_T]) -> Iterator[Type[_T]]:
174 seen: Set[Any] = set()
176 stack = [cls]
177 while stack:
178 cls = stack.pop()
179 if cls in seen:
180 continue
181 else:
182 seen.add(cls)
183 stack.extend(cls.__subclasses__())
184 yield cls
187def string_or_unprintable(element: Any) -> str:
188 if isinstance(element, str):
189 return element
190 else:
191 try:
192 return str(element)
193 except Exception:
194 return "unprintable element %r" % element
197def clsname_as_plain_name(
198 cls: Type[Any], use_name: Optional[str] = None
199) -> str:
200 name = use_name or cls.__name__
201 return " ".join(n.lower() for n in re.findall(r"([A-Z][a-z]+|SQL)", name))
204def method_is_overridden(
205 instance_or_cls: Union[Type[Any], object],
206 against_method: Callable[..., Any],
207) -> bool:
208 """Return True if the two class methods don't match."""
210 if not isinstance(instance_or_cls, type):
211 current_cls = instance_or_cls.__class__
212 else:
213 current_cls = instance_or_cls
215 method_name = against_method.__name__
217 current_method: types.MethodType = getattr(current_cls, method_name)
219 return current_method != against_method
222def decode_slice(slc: slice) -> Tuple[Any, ...]:
223 """decode a slice object as sent to __getitem__.
225 takes into account the 2.5 __index__() method, basically.
227 """
228 ret: List[Any] = []
229 for x in slc.start, slc.stop, slc.step:
230 if hasattr(x, "__index__"):
231 x = x.__index__()
232 ret.append(x)
233 return tuple(ret)
236def _unique_symbols(used: Sequence[str], *bases: str) -> Iterator[str]:
237 used_set = set(used)
238 for base in bases:
239 pool = itertools.chain(
240 (base,),
241 map(lambda i: base + str(i), range(1000)),
242 )
243 for sym in pool:
244 if sym not in used_set:
245 used_set.add(sym)
246 yield sym
247 break
248 else:
249 raise NameError("exhausted namespace for symbol base %s" % base)
252def map_bits(fn: Callable[[int], Any], n: int) -> Iterator[Any]:
253 """Call the given function given each nonzero bit from n."""
255 while n:
256 b = n & (~n + 1)
257 yield fn(b)
258 n ^= b
261_Fn = TypeVar("_Fn", bound="Callable[..., Any]")
263# this seems to be in flux in recent mypy versions
266def decorator(target: Callable[..., Any]) -> Callable[[_Fn], _Fn]:
267 """A signature-matching decorator factory."""
269 def decorate(fn: _Fn) -> _Fn:
270 if not inspect.isfunction(fn) and not inspect.ismethod(fn):
271 raise Exception("not a decoratable function")
273 # Python 3.14 defer creating __annotations__ until its used.
274 # We do not want to create __annotations__ now.
275 annofunc = getattr(fn, "__annotate__", None)
276 if annofunc is not None:
277 fn.__annotate__ = None # type: ignore[union-attr]
278 try:
279 spec = compat.inspect_getfullargspec(fn)
280 finally:
281 fn.__annotate__ = annofunc # type: ignore[union-attr]
282 else:
283 spec = compat.inspect_getfullargspec(fn)
285 # Do not generate code for annotations.
286 # update_wrapper() copies the annotation from fn to decorated.
287 # We use dummy defaults for code generation to avoid having
288 # copy of large globals for compiling.
289 # We copy __defaults__ and __kwdefaults__ from fn to decorated.
290 empty_defaults = (None,) * len(spec.defaults or ())
291 empty_kwdefaults = dict.fromkeys(spec.kwonlydefaults or ())
292 spec = spec._replace(
293 annotations={},
294 defaults=empty_defaults,
295 kwonlydefaults=empty_kwdefaults,
296 )
298 names = (
299 tuple(cast("Tuple[str, ...]", spec[0]))
300 + cast("Tuple[str, ...]", spec[1:3])
301 + (fn.__name__,)
302 )
303 targ_name, fn_name = _unique_symbols(names, "target", "fn")
305 metadata: Dict[str, Optional[str]] = dict(target=targ_name, fn=fn_name)
306 metadata.update(format_argspec_plus(spec, grouped=False))
307 metadata["name"] = fn.__name__
309 if inspect.iscoroutinefunction(fn):
310 metadata["prefix"] = "async "
311 metadata["target_prefix"] = "await "
312 metadata["target_suffix"] = ""
313 elif inspect.isgeneratorfunction(fn):
314 # a generator function has to remain a generator function
315 # after decoration; tools such as pytest fixtures test for
316 # inspect.isgeneratorfunction() and will otherwise never
317 # iterate the function at all
318 metadata["prefix"] = ""
319 metadata["target_prefix"] = "(yield from "
320 metadata["target_suffix"] = ")"
321 else:
322 metadata["prefix"] = ""
323 metadata["target_prefix"] = ""
324 metadata["target_suffix"] = ""
326 # look for __ positional arguments. This is a convention in
327 # SQLAlchemy that arguments should be passed positionally
328 # rather than as keyword
329 # arguments. note that apply_pos doesn't currently work in all cases
330 # such as when a kw-only indicator "*" is present, which is why
331 # we limit the use of this to just that case we can detect. As we add
332 # more kinds of methods that use @decorator, things may have to
333 # be further improved in this area
334 if "__" in repr(spec[0]):
335 code = """\
336%(prefix)sdef %(name)s%(grouped_args)s:
337 return %(target_prefix)s%(target)s(%(fn)s, %(apply_pos)s)%(target_suffix)s
338""" % metadata
339 else:
340 code = """\
341%(prefix)sdef %(name)s%(grouped_args)s:
342 return %(target_prefix)s%(target)s(%(fn)s, %(apply_kw)s)%(target_suffix)s
343""" % metadata
345 env: Dict[str, Any] = {
346 targ_name: target,
347 fn_name: fn,
348 "__name__": fn.__module__,
349 }
351 # the target's name is part of the description because decorators
352 # built here do get stacked (see ValuesBase.values()), and
353 # update_wrapper() gives every layer the same __qualname__; without
354 # it the outer layer would claim the inner layer's source.
355 decorated = cast(
356 types.FunctionType,
357 exec_code_in_env(
358 code,
359 env,
360 fn.__name__,
361 f"{target.__name__}() wrapper for "
362 f"{fn.__module__}.{fn.__qualname__}",
363 ),
364 )
365 decorated.__defaults__ = fn.__defaults__
366 decorated.__kwdefaults__ = fn.__kwdefaults__ # type: ignore[union-attr] # noqa: E501
367 return update_wrapper(decorated, fn) # type: ignore[return-value]
369 return update_wrapper(decorate, target) # type: ignore[return-value]
372_LinecacheEntry = Tuple[int, None, List[str], str]
375def _linecache_cache_getter():
376 """safe getter for linecache.cache
378 linecache.cache despite being non-underscored and widely used is
379 nonetheless not documented by cPython. Therefore we cannot trust that it
380 it's present in third party Python distributions, or that it wont
381 suddenly be removed or changed. Guard against this such that linecache
382 features will be silently disabled if this should happen. Unit tests in
383 test/base/test_utils.py ensures linecache.cache remains available for new
384 releases.
386 """
387 try:
388 linecache_cache = linecache.cache
389 except AttributeError:
390 raise
391 else:
392 if not isinstance(linecache_cache, dict):
393 raise AttributeError("linecache has changed from being a dict")
394 return linecache_cache
397def _remove_linecache_entry(filename: str, entry: _LinecacheEntry) -> None:
398 """Discard a ``linecache`` entry made by :func:`.exec_code_in_env`, if
399 it is still the live one.
401 Runs from a weakref finalizer.
403 This function is only established if we actually added an entry to the
404 linecache within exec_code_in_env.
406 """
407 try:
408 linecache_cache = _linecache_cache_getter()
409 except AttributeError:
410 # linecache.cache is gone or is no longer a dict; as this runs
411 # from a finalizer, an exception here would only surface as an
412 # unraisable error
413 return
415 if linecache_cache.get(filename) is entry:
416 linecache_cache.pop(filename, None)
419def exec_code_in_env(
420 code: Union[str, types.CodeType],
421 env: Dict[str, Any],
422 fn_name: str,
423 description: Optional[str] = None,
424) -> Callable[..., Any]:
425 """Exec generated ``code`` in ``env`` and return the function it defines.
427 If ``description`` is passed, the code is compiled against a synthetic
428 filename which is registered with :mod:`linecache`, allowing traceback
429 frames to reference the actual source code being referenced.
431 entries are placed in the cache without an expiration time and a
432 weakref.finalize() is applied to the function to remove the linecache
433 entry if and when the function is garbage collected.
435 """
436 filename: Optional[str] = None
437 entry: Optional[_LinecacheEntry] = None
439 if description is not None:
440 assert isinstance(code, str), (
441 "a description is only meaningful for source that has not "
442 "already been compiled"
443 )
444 filename = f"<sqlalchemy generated {description}>"
446 try:
447 linecache_cache = _linecache_cache_getter()
448 except AttributeError:
449 pass
450 else:
451 entry = (len(code), None, code.splitlines(True), filename)
452 linecache_cache[filename] = entry
453 code = compile(code, filename, "exec")
455 exec(code, env)
456 fn = env[fn_name]
458 if filename is not None and entry is not None:
459 # apply a finalizer that addresses on-the-fly functions, ORM mapped
460 # classes, etc. which might be garbage collected
461 finalizer = weakref.finalize(
462 fn, _remove_linecache_entry, filename, entry
463 )
464 # atexit is a property on the C implementation; typeshed
465 # renders finalize with an empty __slots__
466 finalizer.atexit = False
468 return fn # type: ignore[no-any-return]
471_PF = TypeVar("_PF")
472_TE = TypeVar("_TE")
475class PluginLoader:
476 def __init__(
477 self, group: str, auto_fn: Optional[Callable[..., Any]] = None
478 ):
479 self.group = group
480 self.impls: Dict[str, Any] = {}
481 self.auto_fn = auto_fn
483 def clear(self):
484 self.impls.clear()
486 def load(self, name: str) -> Any:
487 if name in self.impls:
488 return self.impls[name]()
490 if self.auto_fn:
491 loader = self.auto_fn(name)
492 if loader:
493 self.impls[name] = loader
494 return loader()
496 for impl in compat.importlib_metadata_get(self.group):
497 if impl.name == name:
498 self.impls[name] = impl.load
499 return impl.load()
501 raise exc.NoSuchModuleError(
502 "Can't load plugin: %s:%s" % (self.group, name)
503 )
505 def register(self, name: str, modulepath: str, objname: str) -> None:
506 def load():
507 mod = __import__(modulepath)
508 for token in modulepath.split(".")[1:]:
509 mod = getattr(mod, token)
510 return getattr(mod, objname)
512 self.impls[name] = load
514 def deregister(self, name: str) -> None:
515 del self.impls[name]
518def _inspect_func_args(fn):
519 try:
520 co_varkeywords = inspect.CO_VARKEYWORDS
521 except AttributeError:
522 # https://docs.python.org/3/library/inspect.html
523 # The flags are specific to CPython, and may not be defined in other
524 # Python implementations. Furthermore, the flags are an implementation
525 # detail, and can be removed or deprecated in future Python releases.
526 spec = compat.inspect_getfullargspec(fn)
527 return spec[0], bool(spec[2])
528 else:
529 # use fn.__code__ plus flags to reduce method call overhead
530 co = fn.__code__
531 nargs = co.co_argcount
532 return (
533 list(co.co_varnames[:nargs]),
534 bool(co.co_flags & co_varkeywords),
535 )
538@overload
539def get_cls_kwargs(
540 cls: type,
541 *,
542 _set: Optional[Set[str]] = None,
543 raiseerr: Literal[True] = ...,
544) -> Set[str]: ...
547@overload
548def get_cls_kwargs(
549 cls: type, *, _set: Optional[Set[str]] = None, raiseerr: bool = False
550) -> Optional[Set[str]]: ...
553def get_cls_kwargs(
554 cls: type, *, _set: Optional[Set[str]] = None, raiseerr: bool = False
555) -> Optional[Set[str]]:
556 r"""Return the full set of inherited kwargs for the given `cls`.
558 Probes a class's __init__ method, collecting all named arguments. If the
559 __init__ defines a \**kwargs catch-all, then the constructor is presumed
560 to pass along unrecognized keywords to its base classes, and the
561 collection process is repeated recursively on each of the bases.
563 Uses a subset of inspect.getfullargspec() to cut down on method overhead,
564 as this is used within the Core typing system to create copies of type
565 objects which is a performance-sensitive operation.
567 No anonymous tuple arguments please !
569 """
570 toplevel = _set is None
571 if toplevel:
572 _set = set()
573 assert _set is not None
575 ctr = cls.__dict__.get("__init__", False)
577 has_init = (
578 ctr
579 and isinstance(ctr, types.FunctionType)
580 and isinstance(ctr.__code__, types.CodeType)
581 )
583 if has_init:
584 names, has_kw = _inspect_func_args(ctr)
585 _set.update(names)
587 if not has_kw and not toplevel:
588 if raiseerr:
589 raise TypeError(
590 f"given cls {cls} doesn't have an __init__ method"
591 )
592 else:
593 return None
594 else:
595 has_kw = False
597 if not has_init or has_kw:
598 for c in cls.__bases__:
599 if get_cls_kwargs(c, _set=_set) is None:
600 break
602 _set.discard("self")
603 return _set
606def get_func_kwargs(func: Callable[..., Any]) -> List[str]:
607 """Return the set of legal kwargs for the given `func`.
609 Uses getargspec so is safe to call for methods, functions,
610 etc.
612 """
614 return compat.inspect_getfullargspec(func)[0]
617def get_callable_argspec(
618 fn: Callable[..., Any], no_self: bool = False, _is_init: bool = False
619) -> compat.FullArgSpec:
620 """Return the argument signature for any callable.
622 All pure-Python callables are accepted, including
623 functions, methods, classes, objects with __call__;
624 builtins and other edge cases like functools.partial() objects
625 raise a TypeError.
627 """
628 if inspect.isbuiltin(fn):
629 raise TypeError("Can't inspect builtin: %s" % fn)
630 elif inspect.isfunction(fn) or (
631 hasattr(fn, "__code__")
632 and not inspect.isclass(fn)
633 and not inspect.ismethod(fn)
634 ):
635 if _is_init and no_self:
636 spec = compat.inspect_getfullargspec(fn)
637 return compat.FullArgSpec(
638 spec.args[1:],
639 spec.varargs,
640 spec.varkw,
641 spec.defaults,
642 spec.kwonlyargs,
643 spec.kwonlydefaults,
644 spec.annotations,
645 )
646 else:
647 return compat.inspect_getfullargspec(fn)
648 elif inspect.ismethod(fn):
649 if no_self and (_is_init or fn.__self__):
650 spec = compat.inspect_getfullargspec(fn.__func__)
651 return compat.FullArgSpec(
652 spec.args[1:],
653 spec.varargs,
654 spec.varkw,
655 spec.defaults,
656 spec.kwonlyargs,
657 spec.kwonlydefaults,
658 spec.annotations,
659 )
660 else:
661 return compat.inspect_getfullargspec(fn.__func__)
662 elif inspect.isclass(fn):
663 return get_callable_argspec(
664 fn.__init__, no_self=no_self, _is_init=True
665 )
666 elif hasattr(fn, "__func__"):
667 return compat.inspect_getfullargspec(fn.__func__)
668 elif hasattr(fn, "__call__"):
669 if inspect.ismethod(fn.__call__):
670 return get_callable_argspec(fn.__call__, no_self=no_self)
671 else:
672 raise TypeError("Can't inspect callable: %s" % fn)
673 else:
674 raise TypeError("Can't inspect callable: %s" % fn)
677def format_argspec_plus(
678 fn: Union[Callable[..., Any], compat.FullArgSpec], grouped: bool = True
679) -> Dict[str, Optional[str]]:
680 """Returns a dictionary of formatted, introspected function arguments.
682 A enhanced variant of inspect.formatargspec to support code generation.
684 fn
685 An inspectable callable or tuple of inspect getargspec() results.
686 grouped
687 Defaults to True; include (parens, around, argument) lists
689 Returns:
691 args
692 Full inspect.formatargspec for fn
693 self_arg
694 The name of the first positional argument, varargs[0], or None
695 if the function defines no positional arguments.
696 apply_pos
697 args, re-written in calling rather than receiving syntax. Arguments are
698 passed positionally.
699 apply_kw
700 Like apply_pos, except keyword-ish args are passed as keywords.
701 apply_pos_proxied
702 Like apply_pos but omits the self/cls argument
704 Example::
706 >>> format_argspec_plus(lambda self, a, b, c=3, **d: 123)
707 {'grouped_args': '(self, a, b, c=3, **d)',
708 'self_arg': 'self',
709 'apply_kw': '(self, a, b, c=c, **d)',
710 'apply_pos': '(self, a, b, c, **d)'}
712 """
713 if callable(fn):
714 spec = compat.inspect_getfullargspec(fn)
715 else:
716 spec = fn
718 args = compat.inspect_formatargspec(*spec)
720 apply_pos = compat.inspect_formatargspec(
721 spec[0], spec[1], spec[2], None, spec[4]
722 )
724 if spec[0]:
725 self_arg = spec[0][0]
727 apply_pos_proxied = compat.inspect_formatargspec(
728 spec[0][1:], spec[1], spec[2], None, spec[4]
729 )
731 elif spec[1]:
732 # I'm not sure what this is
733 self_arg = "%s[0]" % spec[1]
735 apply_pos_proxied = apply_pos
736 else:
737 self_arg = None
738 apply_pos_proxied = apply_pos
740 num_defaults = 0
741 if spec[3]:
742 num_defaults += len(cast(Tuple[Any], spec[3]))
743 if spec[4]:
744 num_defaults += len(spec[4])
746 name_args = spec[0] + spec[4]
748 defaulted_vals: Union[List[str], Tuple[()]]
750 if num_defaults:
751 defaulted_vals = name_args[0 - num_defaults :]
752 else:
753 defaulted_vals = ()
755 apply_kw = compat.inspect_formatargspec(
756 name_args,
757 spec[1],
758 spec[2],
759 defaulted_vals,
760 formatvalue=lambda x: "=" + str(x),
761 )
763 if spec[0]:
764 apply_kw_proxied = compat.inspect_formatargspec(
765 name_args[1:],
766 spec[1],
767 spec[2],
768 defaulted_vals,
769 formatvalue=lambda x: "=" + str(x),
770 )
771 else:
772 apply_kw_proxied = apply_kw
774 if grouped:
775 return dict(
776 grouped_args=args,
777 self_arg=self_arg,
778 apply_pos=apply_pos,
779 apply_kw=apply_kw,
780 apply_pos_proxied=apply_pos_proxied,
781 apply_kw_proxied=apply_kw_proxied,
782 )
783 else:
784 return dict(
785 grouped_args=args,
786 self_arg=self_arg,
787 apply_pos=apply_pos[1:-1],
788 apply_kw=apply_kw[1:-1],
789 apply_pos_proxied=apply_pos_proxied[1:-1],
790 apply_kw_proxied=apply_kw_proxied[1:-1],
791 )
794def format_argspec_init(method, grouped=True):
795 """format_argspec_plus with considerations for typical __init__ methods
797 Wraps format_argspec_plus with error handling strategies for typical
798 __init__ cases:
800 .. sourcecode:: text
802 object.__init__ -> (self)
803 other unreflectable (usually C) -> (self, *args, **kwargs)
805 """
806 if method is object.__init__:
807 grouped_args = "(self)"
808 args = "(self)" if grouped else "self"
809 proxied = "()" if grouped else ""
810 else:
811 try:
812 return format_argspec_plus(method, grouped=grouped)
813 except TypeError:
814 grouped_args = "(self, *args, **kwargs)"
815 args = grouped_args if grouped else "self, *args, **kwargs"
816 proxied = "(*args, **kwargs)" if grouped else "*args, **kwargs"
817 return dict(
818 self_arg="self",
819 grouped_args=grouped_args,
820 apply_pos=args,
821 apply_kw=args,
822 apply_pos_proxied=proxied,
823 apply_kw_proxied=proxied,
824 )
827def create_proxy_methods(
828 target_cls: Type[Any],
829 target_cls_sphinx_name: str,
830 proxy_cls_sphinx_name: str,
831 classmethods: Sequence[str] = (),
832 methods: Sequence[str] = (),
833 attributes: Sequence[str] = (),
834 use_intermediate_variable: Sequence[str] = (),
835) -> Callable[[_T], _T]:
836 """A class decorator indicating attributes should refer to a proxy
837 class.
839 This decorator is now a "marker" that does nothing at runtime. Instead,
840 it is consumed by the tools/generate_proxy_methods.py script to
841 statically generate proxy methods and attributes that are fully
842 recognized by typing tools such as mypy.
844 """
846 def decorate(cls):
847 return cls
849 return decorate
852def getargspec_init(method):
853 """inspect.getargspec with considerations for typical __init__ methods
855 Wraps inspect.getargspec with error handling for typical __init__ cases:
857 .. sourcecode:: text
859 object.__init__ -> (self)
860 other unreflectable (usually C) -> (self, *args, **kwargs)
862 """
863 try:
864 return compat.inspect_getfullargspec(method)
865 except TypeError:
866 if method is object.__init__:
867 return (["self"], None, None, None)
868 else:
869 return (["self"], "args", "kwargs", None)
872def unbound_method_to_callable(func_or_cls):
873 """Adjust the incoming callable such that a 'self' argument is not
874 required.
876 """
878 if isinstance(func_or_cls, types.MethodType) and not func_or_cls.__self__:
879 return func_or_cls.__func__
880 else:
881 return func_or_cls
884class GenericRepr:
885 """Encapsulates the logic for creating a generic __repr__() string.
887 This class allows for the repr structure to be created, then modified
888 (e.g., changing the class name), before being rendered as a string.
890 .. versionadded:: 2.1
891 """
893 __slots__ = (
894 "_obj",
895 "_additional_kw",
896 "_to_inspect",
897 "_omit_kwarg",
898 "_class_name",
899 )
901 _obj: Any
902 _additional_kw: Sequence[Tuple[str, Any]]
903 _to_inspect: List[object]
904 _omit_kwarg: Sequence[str]
905 _class_name: Optional[str]
907 def __init__(
908 self,
909 obj: Any,
910 additional_kw: Sequence[Tuple[str, Any]] = (),
911 to_inspect: Optional[Union[object, List[object]]] = None,
912 omit_kwarg: Sequence[str] = (),
913 ):
914 """Create a GenericRepr object.
916 :param obj: The object being repr'd
917 :param additional_kw: Additional keyword arguments to check for in
918 the repr, as a sequence of 2-tuples of (name, default_value)
919 :param to_inspect: One or more objects whose __init__ signature
920 should be inspected. If not provided, defaults to [obj].
921 :param omit_kwarg: Sequence of keyword argument names to omit from
922 the repr output
923 """
924 self._obj = obj
925 self._additional_kw = additional_kw
926 self._to_inspect = (
927 [obj] if to_inspect is None else _collections.to_list(to_inspect)
928 )
929 self._omit_kwarg = omit_kwarg
930 self._class_name = None
932 def set_class_name(self, class_name: str) -> GenericRepr:
933 """Set the class name to be used in the repr.
935 By default, the class name is taken from obj.__class__.__name__.
936 This method allows it to be overridden.
938 :param class_name: The class name to use
939 :return: self, for method chaining
940 """
941 self._class_name = class_name
942 return self
944 def __str__(self) -> str:
945 """Produce the __repr__() string based on the configured parameters."""
946 obj = self._obj
947 to_inspect = self._to_inspect
948 additional_kw = self._additional_kw
949 omit_kwarg = self._omit_kwarg
951 missing = object()
953 pos_args = []
954 kw_args: _collections.OrderedDict[str, Any] = (
955 _collections.OrderedDict()
956 )
957 vargs = None
958 for i, insp in enumerate(to_inspect):
959 try:
960 spec = compat.inspect_getfullargspec(insp.__init__) # type: ignore[misc] # noqa: E501
961 except TypeError:
962 continue
963 else:
964 default_len = len(spec.defaults) if spec.defaults else 0
965 if i == 0:
966 if spec.varargs:
967 vargs = spec.varargs
968 if default_len:
969 pos_args.extend(spec.args[1:-default_len])
970 else:
971 pos_args.extend(spec.args[1:])
972 else:
973 kw_args.update(
974 [(arg, missing) for arg in spec.args[1:-default_len]]
975 )
977 if default_len:
978 assert spec.defaults
979 kw_args.update(
980 [
981 (arg, default)
982 for arg, default in zip(
983 spec.args[-default_len:], spec.defaults
984 )
985 ]
986 )
987 output: List[str] = []
989 output.extend(repr(getattr(obj, arg, None)) for arg in pos_args)
991 if vargs is not None and hasattr(obj, vargs):
992 output.extend([repr(val) for val in getattr(obj, vargs)])
994 for arg, defval in kw_args.items():
995 if arg in omit_kwarg:
996 continue
997 try:
998 val = getattr(obj, arg, missing)
999 if val is not missing and val != defval:
1000 output.append("%s=%r" % (arg, val))
1001 except Exception:
1002 pass
1004 if additional_kw:
1005 for arg, defval in additional_kw:
1006 try:
1007 val = getattr(obj, arg, missing)
1008 if val is not missing and val != defval:
1009 output.append("%s=%r" % (arg, val))
1010 except Exception:
1011 pass
1013 class_name = (
1014 self._class_name
1015 if self._class_name is not None
1016 else obj.__class__.__name__
1017 )
1018 return "%s(%s)" % (class_name, ", ".join(output))
1021def generic_repr(
1022 obj: Any,
1023 additional_kw: Sequence[Tuple[str, Any]] = (),
1024 to_inspect: Optional[Union[object, List[object]]] = None,
1025 omit_kwarg: Sequence[str] = (),
1026) -> str:
1027 """Produce a __repr__() based on direct association of the __init__()
1028 specification vs. same-named attributes present.
1030 """
1031 return str(
1032 GenericRepr(
1033 obj,
1034 additional_kw=additional_kw,
1035 to_inspect=to_inspect,
1036 omit_kwarg=omit_kwarg,
1037 )
1038 )
1041def class_hierarchy(cls):
1042 """Return an unordered sequence of all classes related to cls.
1044 Traverses diamond hierarchies.
1046 Fibs slightly: subclasses of builtin types are not returned. Thus
1047 class_hierarchy(class A(object)) returns (A, object), not A plus every
1048 class systemwide that derives from object.
1049 Likewise, class_hierarchy(class A(Generic[T])) returns
1050 (A, Generic, object), not every class systemwide that derives from Generic.
1052 """
1054 hier = {cls}
1055 process = list(cls.__mro__)
1056 while process:
1057 c = process.pop()
1058 bases = (_ for _ in c.__bases__ if _ not in hier)
1060 for b in bases:
1061 process.append(b)
1062 hier.add(b)
1064 if (
1065 c is Generic
1066 or c.__module__ == "builtins"
1067 or not hasattr(c, "__subclasses__")
1068 ):
1069 continue
1071 for s in [
1072 _
1073 for _ in (
1074 c.__subclasses__()
1075 if not issubclass(c, type)
1076 else c.__subclasses__(c)
1077 )
1078 if _ not in hier
1079 ]:
1080 process.append(s)
1081 hier.add(s)
1082 return list(hier)
1085def iterate_attributes(cls):
1086 """iterate all the keys and attributes associated
1087 with a class, without using getattr().
1089 Does not use getattr() so that class-sensitive
1090 descriptors (i.e. property.__get__()) are not called.
1092 """
1093 keys = dir(cls)
1094 for key in keys:
1095 for c in cls.__mro__:
1096 if key in c.__dict__:
1097 yield (key, c.__dict__[key])
1098 break
1101def monkeypatch_proxied_specials(
1102 into_cls,
1103 from_cls,
1104 skip=None,
1105 only=None,
1106 name="self.proxy",
1107 from_instance=None,
1108):
1109 """Automates delegation of __specials__ for a proxying type."""
1111 if only:
1112 dunders = only
1113 else:
1114 if skip is None:
1115 skip = (
1116 "__slots__",
1117 "__del__",
1118 "__getattribute__",
1119 "__metaclass__",
1120 "__getstate__",
1121 "__setstate__",
1122 )
1123 dunders = [
1124 m
1125 for m in dir(from_cls)
1126 if (
1127 m.startswith("__")
1128 and m.endswith("__")
1129 and not hasattr(into_cls, m)
1130 and m not in skip
1131 )
1132 ]
1134 for method in dunders:
1135 try:
1136 maybe_fn = getattr(from_cls, method)
1137 if not hasattr(maybe_fn, "__call__"):
1138 continue
1139 maybe_fn = getattr(maybe_fn, "__func__", maybe_fn)
1140 fn = cast(types.FunctionType, maybe_fn)
1142 except AttributeError:
1143 continue
1144 try:
1145 spec = compat.inspect_getfullargspec(fn)
1146 fn_args = compat.inspect_formatargspec(spec[0])
1147 d_args = compat.inspect_formatargspec(spec[0][1:])
1148 except TypeError:
1149 fn_args = "(self, *args, **kw)"
1150 d_args = "(*args, **kw)"
1152 py = (
1153 "def %(method)s%(fn_args)s: "
1154 "return %(name)s.%(method)s%(d_args)s" % locals()
1155 )
1157 env: Dict[str, types.FunctionType] = (
1158 from_instance is not None and {name: from_instance} or {}
1159 )
1160 # the generated source is derived entirely from from_cls and the method
1161 # name, and into_cls is a throwaway per-descriptor class whose
1162 # __qualname__ is the same for every one of them, so this is keyed on
1163 # from_cls rather than into_cls
1164 proxied = exec_code_in_env(
1165 py,
1166 env,
1167 method,
1168 f"{method} proxying to "
1169 f"{from_cls.__module__}.{from_cls.__qualname__}",
1170 )
1171 try:
1172 proxied.__defaults__ = fn.__defaults__
1173 except AttributeError:
1174 pass
1175 setattr(into_cls, method, proxied)
1178def methods_equivalent(meth1, meth2):
1179 """Return True if the two methods are the same implementation."""
1181 return getattr(meth1, "__func__", meth1) is getattr(
1182 meth2, "__func__", meth2
1183 )
1186def as_interface(obj, cls=None, methods=None, required=None):
1187 """Ensure basic interface compliance for an instance or dict of callables.
1189 Checks that ``obj`` implements public methods of ``cls`` or has members
1190 listed in ``methods``. If ``required`` is not supplied, implementing at
1191 least one interface method is sufficient. Methods present on ``obj`` that
1192 are not in the interface are ignored.
1194 If ``obj`` is a dict and ``dict`` does not meet the interface
1195 requirements, the keys of the dictionary are inspected. Keys present in
1196 ``obj`` that are not in the interface will raise TypeErrors.
1198 Raises TypeError if ``obj`` does not meet the interface criteria.
1200 In all passing cases, an object with callable members is returned. In the
1201 simple case, ``obj`` is returned as-is; if dict processing kicks in then
1202 an anonymous class is returned.
1204 obj
1205 A type, instance, or dictionary of callables.
1206 cls
1207 Optional, a type. All public methods of cls are considered the
1208 interface. An ``obj`` instance of cls will always pass, ignoring
1209 ``required``..
1210 methods
1211 Optional, a sequence of method names to consider as the interface.
1212 required
1213 Optional, a sequence of mandatory implementations. If omitted, an
1214 ``obj`` that provides at least one interface method is considered
1215 sufficient. As a convenience, required may be a type, in which case
1216 all public methods of the type are required.
1218 """
1219 if not cls and not methods:
1220 raise TypeError("a class or collection of method names are required")
1222 if isinstance(cls, type) and isinstance(obj, cls):
1223 return obj
1225 interface = set(methods or [m for m in dir(cls) if not m.startswith("_")])
1226 implemented = set(dir(obj))
1228 complies = operator.ge
1229 if isinstance(required, type):
1230 required = interface
1231 elif not required:
1232 required = set()
1233 complies = operator.gt
1234 else:
1235 required = set(required)
1237 if complies(implemented.intersection(interface), required):
1238 return obj
1240 # No dict duck typing here.
1241 if not isinstance(obj, dict):
1242 qualifier = complies is operator.gt and "any of" or "all of"
1243 raise TypeError(
1244 "%r does not implement %s: %s"
1245 % (obj, qualifier, ", ".join(interface))
1246 )
1248 class AnonymousInterface:
1249 """A callable-holding shell."""
1251 if cls:
1252 AnonymousInterface.__name__ = "Anonymous" + cls.__name__
1253 found = set()
1255 for method, impl in dictlike_iteritems(obj):
1256 if method not in interface:
1257 raise TypeError("%r: unknown in this interface" % method)
1258 if not callable(impl):
1259 raise TypeError("%r=%r is not callable" % (method, impl))
1260 setattr(AnonymousInterface, method, staticmethod(impl))
1261 found.add(method)
1263 if complies(found, required):
1264 return AnonymousInterface
1266 raise TypeError(
1267 "dictionary does not contain required keys %s"
1268 % ", ".join(required - found)
1269 )
1272_GFD = TypeVar("_GFD", bound="generic_fn_descriptor[Any]")
1275class generic_fn_descriptor(Generic[_T_co]):
1276 """Descriptor which proxies a function when the attribute is not
1277 present in dict
1279 This superclass is organized in a particular way with "memoized" and
1280 "non-memoized" implementation classes that are hidden from type checkers,
1281 as Mypy seems to not be able to handle seeing multiple kinds of descriptor
1282 classes used for the same attribute.
1284 """
1286 fget: Callable[..., _T_co]
1287 __doc__: Optional[str]
1288 __name__: str
1290 def __init__(self, fget: Callable[..., _T_co], doc: Optional[str] = None):
1291 self.fget = fget
1292 self.__doc__ = doc or fget.__doc__
1293 self.__name__ = fget.__name__
1295 @overload
1296 def __get__(self: _GFD, obj: None, cls: Any) -> _GFD: ...
1298 @overload
1299 def __get__(self, obj: object, cls: Any) -> _T_co: ...
1301 def __get__(self: _GFD, obj: Any, cls: Any) -> Union[_GFD, _T_co]:
1302 raise NotImplementedError()
1304 if TYPE_CHECKING:
1306 def __set__(self, instance: Any, value: Any) -> None: ...
1308 def __delete__(self, instance: Any) -> None: ...
1310 def _reset(self, obj: Any) -> None:
1311 raise NotImplementedError()
1313 @classmethod
1314 def reset(cls, obj: Any, name: str) -> None:
1315 raise NotImplementedError()
1318class _non_memoized_property(generic_fn_descriptor[_T_co]):
1319 """a plain descriptor that proxies a function.
1321 primary rationale is to provide a plain attribute that's
1322 compatible with memoized_property which is also recognized as equivalent
1323 by mypy.
1325 """
1327 if not TYPE_CHECKING:
1329 def __get__(self, obj, cls):
1330 if obj is None:
1331 return self
1332 return self.fget(obj)
1335class _memoized_property(generic_fn_descriptor[_T_co]):
1336 """A read-only @property that is only evaluated once."""
1338 if not TYPE_CHECKING:
1340 def __get__(self, obj, cls):
1341 if obj is None:
1342 return self
1343 obj.__dict__[self.__name__] = result = self.fget(obj)
1344 return result
1346 def _reset(self, obj):
1347 _memoized_property.reset(obj, self.__name__)
1349 @classmethod
1350 def reset(cls, obj, name):
1351 obj.__dict__.pop(name, None)
1354# despite many attempts to get Mypy to recognize an overridden descriptor
1355# where one is memoized and the other isn't, there seems to be no reliable
1356# way other than completely deceiving the type checker into thinking there
1357# is just one single descriptor type everywhere. Otherwise, if a superclass
1358# has non-memoized and subclass has memoized, that requires
1359# "class memoized(non_memoized)". but then if a superclass has memoized and
1360# superclass has non-memoized, the class hierarchy of the descriptors
1361# would need to be reversed; "class non_memoized(memoized)". so there's no
1362# way to achieve this.
1363# additional issues, RO properties:
1364# https://github.com/python/mypy/issues/12440
1365if TYPE_CHECKING:
1366 # allow memoized and non-memoized to be freely mixed by having them
1367 # be the same class
1368 memoized_property = generic_fn_descriptor
1369 non_memoized_property = generic_fn_descriptor
1371 # for read only situations, mypy only sees @property as read only.
1372 # read only is needed when a subtype specializes the return type
1373 # of a property, meaning assignment needs to be disallowed
1374 ro_memoized_property = property
1375 ro_non_memoized_property = property
1377else:
1378 memoized_property = ro_memoized_property = _memoized_property
1379 non_memoized_property = ro_non_memoized_property = _non_memoized_property
1382def memoized_instancemethod(fn: _F) -> _F:
1383 """Decorate a method memoize its return value.
1385 Best applied to no-arg methods: memoization is not sensitive to
1386 argument values, and will always return the same value even when
1387 called with different arguments.
1389 """
1391 def oneshot(self, *args, **kw):
1392 result = fn(self, *args, **kw)
1394 def memo(*a, **kw):
1395 return result
1397 memo.__name__ = fn.__name__
1398 memo.__doc__ = fn.__doc__
1399 self.__dict__[fn.__name__] = memo
1400 return result
1402 return update_wrapper(oneshot, fn) # type: ignore[return-value]
1405class HasMemoized:
1406 """A mixin class that maintains the names of memoized elements in a
1407 collection for easy cache clearing, generative, etc.
1409 """
1411 if not TYPE_CHECKING:
1412 # support classes that want to have __slots__ with an explicit
1413 # slot for __dict__. not sure if that requires base __slots__ here.
1414 __slots__ = ()
1416 _memoized_keys: FrozenSet[str] = frozenset()
1418 def _reset_memoizations(self) -> None:
1419 for elem in self._memoized_keys:
1420 self.__dict__.pop(elem, None)
1422 def _assert_no_memoizations(self) -> None:
1423 for elem in self._memoized_keys:
1424 assert elem not in self.__dict__
1426 def _set_memoized_attribute(self, key: str, value: Any) -> None:
1427 self.__dict__[key] = value
1428 self._memoized_keys |= {key}
1430 class memoized_attribute(memoized_property[_T]):
1431 """A read-only @property that is only evaluated once.
1433 :meta private:
1435 """
1437 fget: Callable[..., _T]
1438 __doc__: Optional[str]
1439 __name__: str
1441 def __init__(self, fget: Callable[..., _T], doc: Optional[str] = None):
1442 self.fget = fget
1443 self.__doc__ = doc or fget.__doc__
1444 self.__name__ = fget.__name__
1446 @overload
1447 def __get__(self: _MA, obj: None, cls: Any) -> _MA: ...
1449 @overload
1450 def __get__(self, obj: Any, cls: Any) -> _T: ...
1452 def __get__(self, obj, cls):
1453 if obj is None:
1454 return self
1455 obj.__dict__[self.__name__] = result = self.fget(obj)
1456 obj._memoized_keys |= {self.__name__}
1457 return result
1459 @classmethod
1460 def memoized_instancemethod(cls, fn: _F) -> _F:
1461 """Decorate a method memoize its return value.
1463 :meta private:
1465 """
1467 def oneshot(self: Any, *args: Any, **kw: Any) -> Any:
1468 result = fn(self, *args, **kw)
1470 def memo(*a, **kw):
1471 return result
1473 memo.__name__ = fn.__name__
1474 memo.__doc__ = fn.__doc__
1475 self.__dict__[fn.__name__] = memo
1476 self._memoized_keys |= {fn.__name__}
1477 return result
1479 return update_wrapper(oneshot, fn) # type: ignore[return-value]
1482if TYPE_CHECKING:
1483 HasMemoized_ro_memoized_attribute = property
1484else:
1485 HasMemoized_ro_memoized_attribute = HasMemoized.memoized_attribute
1488class MemoizedSlots:
1489 """Apply memoized items to an object using a __getattr__ scheme.
1491 This allows the functionality of memoized_property and
1492 memoized_instancemethod to be available to a class using __slots__.
1494 The memoized get is not threadsafe under freethreading and the
1495 creator method may in extremely rare cases be called more than once.
1497 """
1499 __slots__ = ()
1501 def _fallback_getattr(self, key):
1502 raise AttributeError(key)
1504 def __getattr__(self, key: str) -> Any:
1505 if key.startswith("_memoized_attr_") or key.startswith(
1506 "_memoized_method_"
1507 ):
1508 raise AttributeError(key)
1509 # to avoid recursion errors when interacting with other __getattr__
1510 # schemes that refer to this one, when testing for memoized method
1511 # look at __class__ only rather than going into __getattr__ again.
1512 elif hasattr(self.__class__, f"_memoized_attr_{key}"):
1513 value = getattr(self, f"_memoized_attr_{key}")()
1514 setattr(self, key, value)
1515 return value
1516 elif hasattr(self.__class__, f"_memoized_method_{key}"):
1517 meth = getattr(self, f"_memoized_method_{key}")
1519 def oneshot(*args, **kw):
1520 result = meth(*args, **kw)
1522 def memo(*a, **kw):
1523 return result
1525 memo.__name__ = meth.__name__
1526 memo.__doc__ = meth.__doc__
1527 setattr(self, key, memo)
1528 return result
1530 oneshot.__doc__ = meth.__doc__
1531 return oneshot
1532 else:
1533 return self._fallback_getattr(key)
1536# from paste.deploy.converters
1537def asbool(obj: Any) -> bool:
1538 if isinstance(obj, str):
1539 obj = obj.strip().lower()
1540 if obj in ["true", "yes", "on", "y", "t", "1"]:
1541 return True
1542 elif obj in ["false", "no", "off", "n", "f", "0"]:
1543 return False
1544 else:
1545 raise ValueError("String is not true/false: %r" % obj)
1546 return bool(obj)
1549def bool_or_str(*text: str) -> Callable[[str], Union[str, bool]]:
1550 """Return a callable that will evaluate a string as
1551 boolean, or one of a set of "alternate" string values.
1553 """
1555 def bool_or_value(obj: str) -> Union[str, bool]:
1556 if obj in text:
1557 return obj
1558 else:
1559 return asbool(obj)
1561 return bool_or_value
1564def asint(value: Any) -> Optional[int]:
1565 """Coerce to integer."""
1567 if value is None:
1568 return value
1569 return int(value)
1572def coerce_kw_type(
1573 kw: Dict[str, Any],
1574 key: str,
1575 type_: Type[Any],
1576 flexi_bool: bool = True,
1577 dest: Optional[Dict[str, Any]] = None,
1578) -> None:
1579 r"""If 'key' is present in dict 'kw', coerce its value to type 'type\_' if
1580 necessary. If 'flexi_bool' is True, the string '0' is considered false
1581 when coercing to boolean.
1582 """
1584 if dest is None:
1585 dest = kw
1587 if (
1588 key in kw
1589 and (not isinstance(type_, type) or not isinstance(kw[key], type_))
1590 and kw[key] is not None
1591 ):
1592 if type_ is bool and flexi_bool:
1593 dest[key] = asbool(kw[key])
1594 else:
1595 dest[key] = type_(kw[key])
1598def constructor_key(obj: Any, cls: Type[Any]) -> Tuple[Any, ...]:
1599 """Produce a tuple structure that is cacheable using the __dict__ of
1600 obj to retrieve values
1602 """
1603 names = get_cls_kwargs(cls)
1604 return (cls,) + tuple(
1605 (k, obj.__dict__[k]) for k in names if k in obj.__dict__
1606 )
1609def constructor_copy(obj: _T, cls: Type[_T], *args: Any, **kw: Any) -> _T:
1610 """Instantiate cls using the __dict__ of obj as constructor arguments.
1612 Uses inspect to match the named arguments of ``cls``.
1614 """
1616 names = get_cls_kwargs(cls)
1617 kw.update(
1618 (k, obj.__dict__[k]) for k in names.difference(kw) if k in obj.__dict__
1619 )
1620 return cls(*args, **kw)
1623def counter() -> Callable[[], int]:
1624 """Return a threadsafe counter function."""
1626 lock = threading.Lock()
1627 counter = itertools.count(1)
1629 # avoid the 2to3 "next" transformation...
1630 def _next():
1631 with lock:
1632 return next(counter)
1634 return _next
1637def duck_type_collection(
1638 specimen: Any, default: Optional[Type[Any]] = None
1639) -> Optional[Type[Any]]:
1640 """Given an instance or class, guess if it is or is acting as one of
1641 the basic collection types: list, set and dict. If the __emulates__
1642 property is present, return that preferentially.
1643 """
1645 if hasattr(specimen, "__emulates__"):
1646 # canonicalize set vs sets.Set to a standard: the builtin set
1647 if specimen.__emulates__ is not None and issubclass(
1648 specimen.__emulates__, set
1649 ):
1650 return set
1651 else:
1652 return specimen.__emulates__ # type: ignore[no-any-return]
1654 isa = issubclass if isinstance(specimen, type) else isinstance
1655 if isa(specimen, list):
1656 return list
1657 elif isa(specimen, set):
1658 return set
1659 elif isa(specimen, dict):
1660 return dict
1662 if hasattr(specimen, "append"):
1663 return list
1664 elif hasattr(specimen, "add"):
1665 return set
1666 elif hasattr(specimen, "set"):
1667 return dict
1668 else:
1669 return default
1672def assert_arg_type(
1673 arg: Any, argtype: Union[Tuple[Type[Any], ...], Type[Any]], name: str
1674) -> Any:
1675 if isinstance(arg, argtype):
1676 return arg
1677 else:
1678 if isinstance(argtype, tuple):
1679 raise exc.ArgumentError(
1680 "Argument '%s' is expected to be one of type %s, got '%s'"
1681 % (name, " or ".join("'%s'" % a for a in argtype), type(arg))
1682 )
1683 else:
1684 raise exc.ArgumentError(
1685 "Argument '%s' is expected to be of type '%s', got '%s'"
1686 % (name, argtype, type(arg))
1687 )
1690def dictlike_iteritems(dictlike):
1691 """Return a (key, value) iterator for almost any dict-like object."""
1693 if hasattr(dictlike, "items"):
1694 return list(dictlike.items())
1696 getter = getattr(dictlike, "__getitem__", getattr(dictlike, "get", None))
1697 if getter is None:
1698 raise TypeError("Object '%r' is not dict-like" % dictlike)
1700 if hasattr(dictlike, "iterkeys"):
1702 def iterator():
1703 for key in dictlike.iterkeys():
1704 assert getter is not None
1705 yield key, getter(key)
1707 return iterator()
1708 elif hasattr(dictlike, "keys"):
1709 return iter((key, getter(key)) for key in dictlike.keys())
1710 else:
1711 raise TypeError("Object '%r' is not dict-like" % dictlike)
1714class classproperty(property):
1715 """A decorator that behaves like @property except that operates
1716 on classes rather than instances.
1718 The decorator is currently special when using the declarative
1719 module, but note that the
1720 :class:`~.sqlalchemy.ext.declarative.declared_attr`
1721 decorator should be used for this purpose with declarative.
1723 """
1725 fget: Callable[[Any], Any]
1727 def __init__(self, fget: Callable[[Any], Any], *arg: Any, **kw: Any):
1728 super().__init__(fget, *arg, **kw)
1729 self.__doc__ = fget.__doc__
1731 def __get__(self, obj: Any, cls: Optional[type] = None) -> Any:
1732 return self.fget(cls)
1735class hybridproperty(Generic[_T]):
1736 def __init__(self, func: Callable[..., _T]):
1737 self.func = func
1738 self.clslevel = func
1740 def __get__(self, instance: Any, owner: Any) -> _T:
1741 if instance is None:
1742 clsval = self.clslevel(owner)
1743 return clsval
1744 else:
1745 return self.func(instance)
1747 def classlevel(self, func: Callable[..., Any]) -> hybridproperty[_T]:
1748 self.clslevel = func
1749 return self
1752class rw_hybridproperty(Generic[_T]):
1753 def __init__(self, func: Callable[..., _T]):
1754 self.func = func
1755 self.clslevel = func
1756 self.setfn: Optional[Callable[..., Any]] = None
1758 def __get__(self, instance: Any, owner: Any) -> _T:
1759 if instance is None:
1760 clsval = self.clslevel(owner)
1761 return clsval
1762 else:
1763 return self.func(instance)
1765 def __set__(self, instance: Any, value: Any) -> None:
1766 assert self.setfn is not None
1767 self.setfn(instance, value)
1769 def setter(self, func: Callable[..., Any]) -> rw_hybridproperty[_T]:
1770 self.setfn = func
1771 return self
1773 def classlevel(self, func: Callable[..., Any]) -> rw_hybridproperty[_T]:
1774 self.clslevel = func
1775 return self
1778class hybridmethod(Generic[_T]):
1779 """Decorate a function as cls- or instance- level."""
1781 def __init__(self, func: Callable[..., _T]):
1782 self.func = self.__func__ = func
1783 self.clslevel = func
1785 def __get__(self, instance: Any, owner: Any) -> Callable[..., _T]:
1786 if instance is None:
1787 return self.clslevel.__get__( # type: ignore[no-any-return]
1788 owner, owner.__class__
1789 )
1790 else:
1791 return self.func.__get__( # type: ignore[no-any-return]
1792 instance, owner
1793 )
1795 def classlevel(self, func: Callable[..., Any]) -> hybridmethod[_T]:
1796 self.clslevel = func
1797 return self
1800class symbol(int):
1801 """A constant symbol.
1803 >>> symbol("foo") is symbol("foo")
1804 True
1805 >>> symbol("foo")
1806 <symbol 'foo>
1808 A slight refinement of the MAGICCOOKIE=object() pattern. The primary
1809 advantage of symbol() is its repr(). They are also singletons.
1811 Repeated calls of symbol('name') will all return the same instance.
1813 """
1815 name: str
1817 symbols: Dict[str, symbol] = {}
1818 _lock = threading.Lock()
1820 def __new__(
1821 cls,
1822 name: str,
1823 doc: Optional[str] = None,
1824 canonical: Optional[int] = None,
1825 ) -> symbol:
1826 with cls._lock:
1827 sym = cls.symbols.get(name)
1828 if sym is None:
1829 assert isinstance(name, str)
1830 if canonical is None:
1831 canonical = hash(name)
1832 sym = int.__new__(symbol, canonical)
1833 sym.name = name
1834 if doc:
1835 sym.__doc__ = doc
1837 # NOTE: we should ultimately get rid of this global thing,
1838 # however, currently it is to support pickling. The best
1839 # change would be when we are on py3.11 at a minimum, we
1840 # switch to stdlib enum.IntFlag.
1841 cls.symbols[name] = sym
1842 else:
1843 if canonical and canonical != sym:
1844 raise TypeError(
1845 f"Can't replace canonical symbol for {name!r} "
1846 f"with new int value {canonical}"
1847 )
1848 return sym
1850 def __reduce__(self):
1851 return symbol, (self.name, "x", int(self))
1853 def __str__(self):
1854 return repr(self)
1856 def __repr__(self):
1857 return f"symbol({self.name!r})"
1860class _IntFlagMeta(type):
1861 def __init__(
1862 cls,
1863 classname: str,
1864 bases: Tuple[Type[Any], ...],
1865 dict_: Dict[str, Any],
1866 **kw: Any,
1867 ) -> None:
1868 items: List[symbol]
1869 cls._items = items = []
1870 for k, v in dict_.items():
1871 if re.match(r"^__.*__$", k):
1872 continue
1873 if isinstance(v, int):
1874 sym = symbol(k, canonical=v)
1875 elif not k.startswith("_"):
1876 raise TypeError("Expected integer values for IntFlag")
1877 else:
1878 continue
1879 setattr(cls, k, sym)
1880 items.append(sym)
1882 cls.__members__ = _collections.immutabledict(
1883 {sym.name: sym for sym in items}
1884 )
1886 def __iter__(self) -> Iterator[symbol]:
1887 raise NotImplementedError(
1888 "iter not implemented to ensure compatibility with "
1889 "Python 3.11 IntFlag. Please use __members__. See "
1890 "https://github.com/python/cpython/issues/99304"
1891 )
1894class _FastIntFlag(metaclass=_IntFlagMeta):
1895 """An 'IntFlag' copycat that isn't slow when performing bitwise
1896 operations.
1898 the ``FastIntFlag`` class will return ``enum.IntFlag`` under TYPE_CHECKING
1899 and ``_FastIntFlag`` otherwise.
1901 """
1904if TYPE_CHECKING:
1905 from enum import IntFlag
1907 FastIntFlag = IntFlag
1908else:
1909 FastIntFlag = _FastIntFlag
1912_E = TypeVar("_E", bound=enum.Enum)
1915def parse_user_argument_for_enum(
1916 arg: Any,
1917 choices: Dict[_E, List[Any]],
1918 name: str,
1919 resolve_symbol_names: bool = False,
1920) -> Optional[_E]:
1921 """Given a user parameter, parse the parameter into a chosen value
1922 from a list of choice objects, typically Enum values.
1924 The user argument can be a string name that matches the name of a
1925 symbol, or the symbol object itself, or any number of alternate choices
1926 such as True/False/ None etc.
1928 :param arg: the user argument.
1929 :param choices: dictionary of enum values to lists of possible
1930 entries for each.
1931 :param name: name of the argument. Used in an :class:`.ArgumentError`
1932 that is raised if the parameter doesn't match any available argument.
1934 """
1935 for enum_value, choice in choices.items():
1936 if arg is enum_value:
1937 return enum_value
1938 elif resolve_symbol_names and arg == enum_value.name:
1939 return enum_value
1940 elif arg in choice:
1941 return enum_value
1943 if arg is None:
1944 return None
1946 raise exc.ArgumentError(f"Invalid value for '{name}': {arg!r}")
1949_creation_order = 1
1952def set_creation_order(instance: Any) -> None:
1953 """Assign a '_creation_order' sequence to the given instance.
1955 This allows multiple instances to be sorted in order of creation
1956 (typically within a single thread; the counter is not particularly
1957 threadsafe).
1959 """
1960 global _creation_order
1961 instance._creation_order = _creation_order
1962 _creation_order += 1
1965def warn_exception(func: Callable[..., Any], *args: Any, **kwargs: Any) -> Any:
1966 """executes the given function, catches all exceptions and converts to
1967 a warning.
1969 """
1970 try:
1971 return func(*args, **kwargs)
1972 except Exception:
1973 warn("%s('%s') ignored" % sys.exc_info()[0:2])
1976def ellipses_string(value, len_=25):
1977 try:
1978 if len(value) > len_:
1979 return "%s..." % value[0:len_]
1980 else:
1981 return value
1982 except TypeError:
1983 return value
1986class _hash_limit_string(str):
1987 """A string subclass that can only be hashed on a maximum amount
1988 of unique values.
1990 This is used for warnings so that we can send out parameterized warnings
1991 without the __warningregistry__ of the module, or the non-overridable
1992 "once" registry within warnings.py, overloading memory,
1995 """
1997 _hash: int
1999 def __new__(
2000 cls, value: str, num: int, args: Sequence[Any]
2001 ) -> _hash_limit_string:
2002 interpolated = (value % args) + (
2003 " (this warning may be suppressed after %d occurrences)" % num
2004 )
2005 self = super().__new__(cls, interpolated)
2006 self._hash = hash("%s_%d" % (value, hash(interpolated) % num))
2007 return self
2009 def __hash__(self) -> int:
2010 return self._hash
2012 def __eq__(self, other: Any) -> bool:
2013 return hash(self) == hash(other)
2016def warn(msg: str, code: Optional[str] = None) -> None:
2017 """Issue a warning.
2019 If msg is a string, :class:`.exc.SAWarning` is used as
2020 the category.
2022 """
2023 if code:
2024 _warnings_warn(exc.SAWarning(msg, code=code))
2025 else:
2026 _warnings_warn(msg, exc.SAWarning)
2029def warn_limited(msg: str, args: Sequence[Any]) -> None:
2030 """Issue a warning with a parameterized string, limiting the number
2031 of registrations.
2033 """
2034 if args:
2035 msg = _hash_limit_string(msg, 10, args)
2036 _warnings_warn(msg, exc.SAWarning)
2039_warning_tags: Dict[CodeType, Tuple[str, Type[Warning]]] = {}
2042def tag_method_for_warnings(
2043 message: str, category: Type[Warning]
2044) -> Callable[[_F], _F]:
2045 def go(fn):
2046 _warning_tags[fn.__code__] = (message, category)
2047 return fn
2049 return go
2052_not_sa_pattern = re.compile(r"^(?:sqlalchemy\.(?!testing)|alembic\.)")
2055def _warnings_warn(
2056 message: Union[str, Warning],
2057 category: Optional[Type[Warning]] = None,
2058 stacklevel: int = 2,
2059) -> None:
2061 if category is None and isinstance(message, Warning):
2062 category = type(message)
2064 # adjust the given stacklevel to be outside of SQLAlchemy
2065 try:
2066 frame = sys._getframe(stacklevel)
2067 except ValueError:
2068 # being called from less than 3 (or given) stacklevels, weird,
2069 # but don't crash
2070 stacklevel = 0
2071 except:
2072 # _getframe() doesn't work, weird interpreter issue, weird,
2073 # ok, but don't crash
2074 stacklevel = 0
2075 else:
2076 stacklevel_found = warning_tag_found = False
2077 while frame is not None:
2078 # using __name__ here requires that we have __name__ in the
2079 # __globals__ of the decorated string functions we make also.
2080 # we generate this using {"__name__": fn.__module__}
2081 if not stacklevel_found and not re.match(
2082 _not_sa_pattern, frame.f_globals.get("__name__", "")
2083 ):
2084 # stop incrementing stack level if an out-of-SQLA line
2085 # were found.
2086 stacklevel_found = True
2088 # however, for the warning tag thing, we have to keep
2089 # scanning up the whole traceback
2091 if frame.f_code in _warning_tags:
2092 warning_tag_found = True
2093 _suffix, _category = _warning_tags[frame.f_code]
2094 category = category or _category
2095 message = f"{message} ({_suffix})"
2097 frame = frame.f_back # type: ignore[assignment]
2099 if not stacklevel_found:
2100 stacklevel += 1
2101 elif stacklevel_found and warning_tag_found:
2102 break
2104 if category is not None:
2105 warnings.warn(message, category, stacklevel=stacklevel + 1)
2106 else:
2107 warnings.warn(message, stacklevel=stacklevel + 1)
2110def only_once(
2111 fn: Callable[..., _T], retry_on_exception: bool
2112) -> Callable[..., Optional[_T]]:
2113 """Decorate the given function to be a no-op after it is called exactly
2114 once."""
2116 once = [fn]
2118 def go(*arg: Any, **kw: Any) -> Optional[_T]:
2119 # strong reference fn so that it isn't garbage collected,
2120 # which interferes with the event system's expectations
2121 strong_fn = fn # noqa
2122 if once:
2123 once_fn = once.pop()
2124 try:
2125 return once_fn(*arg, **kw)
2126 except:
2127 if retry_on_exception:
2128 once.insert(0, once_fn)
2129 raise
2131 return None
2133 return go
2136_SQLA_RE = re.compile(r"sqlalchemy/([a-z_]+/){0,2}[a-z_]+\.py")
2137_UNITTEST_RE = re.compile(r"unit(?:2|test2?/)")
2140def chop_traceback(
2141 tb: List[str],
2142 exclude_prefix: re.Pattern[str] = _UNITTEST_RE,
2143 exclude_suffix: re.Pattern[str] = _SQLA_RE,
2144) -> List[str]:
2145 """Chop extraneous lines off beginning and end of a traceback.
2147 :param tb:
2148 a list of traceback lines as returned by ``traceback.format_stack()``
2150 :param exclude_prefix:
2151 a regular expression object matching lines to skip at beginning of
2152 ``tb``
2154 :param exclude_suffix:
2155 a regular expression object matching lines to skip at end of ``tb``
2156 """
2157 start = 0
2158 end = len(tb) - 1
2159 while start <= end and exclude_prefix.search(tb[start]):
2160 start += 1
2161 while start <= end and exclude_suffix.search(tb[end]):
2162 end -= 1
2163 return tb[start : end + 1]
2166def attrsetter(attrname):
2167 code = "def set(obj, value): obj.%s = value" % attrname
2168 env = locals().copy()
2169 exec(code, env)
2170 return env["set"]
2173dunders_re = re.compile("^__.+__$")
2176class TypingOnly:
2177 """A mixin class that marks a class as 'typing only', meaning it has
2178 absolutely no methods, attributes, or runtime functionality whatsoever.
2180 """
2182 __slots__ = ()
2184 def __init_subclass__(cls, **kw: Any) -> None:
2185 if TypingOnly in cls.__bases__:
2186 remaining = {
2187 name for name in cls.__dict__ if not dunders_re.match(name)
2188 }
2189 if remaining:
2190 raise AssertionError(
2191 f"Class {cls} directly inherits TypingOnly but has "
2192 f"additional attributes {remaining}."
2193 )
2194 super().__init_subclass__(**kw)
2197class EnsureKWArg:
2198 r"""Apply translation of functions to accept \**kw arguments if they
2199 don't already.
2201 Used to ensure cross-compatibility with third party legacy code, for things
2202 like compiler visit methods that need to accept ``**kw`` arguments,
2203 but may have been copied from old code that didn't accept them.
2205 """
2207 ensure_kwarg: str
2208 """a regular expression that indicates method names for which the method
2209 should accept ``**kw`` arguments.
2211 The class will scan for methods matching the name template and decorate
2212 them if necessary to ensure ``**kw`` parameters are accepted.
2214 """
2216 def __init_subclass__(cls) -> None:
2217 fn_reg = cls.ensure_kwarg
2218 clsdict = cls.__dict__
2219 if fn_reg:
2220 for key in clsdict:
2221 m = re.match(fn_reg, key)
2222 if m:
2223 fn = clsdict[key]
2224 spec = compat.inspect_getfullargspec(fn)
2225 if not spec.varkw:
2226 wrapped = cls._wrap_w_kw(fn)
2227 setattr(cls, key, wrapped)
2228 super().__init_subclass__()
2230 @classmethod
2231 def _wrap_w_kw(cls, fn: Callable[..., Any]) -> Callable[..., Any]:
2232 def wrap(*arg: Any, **kw: Any) -> Any:
2233 return fn(*arg)
2235 return update_wrapper(wrap, fn)
2238def wrap_callable(wrapper, fn):
2239 """Augment functools.update_wrapper() to work with objects with
2240 a ``__call__()`` method.
2242 :param fn:
2243 object with __call__ method
2245 """
2246 if hasattr(fn, "__name__"):
2247 return update_wrapper(wrapper, fn)
2248 else:
2249 _f = wrapper
2250 _f.__name__ = fn.__class__.__name__
2251 if hasattr(fn, "__module__"):
2252 _f.__module__ = fn.__module__
2254 if hasattr(fn.__call__, "__doc__") and fn.__call__.__doc__:
2255 _f.__doc__ = fn.__call__.__doc__
2256 elif fn.__doc__:
2257 _f.__doc__ = fn.__doc__
2259 return _f
2262def find_matching_paren(text: str, start: int = 0) -> Optional[int]:
2263 """Return the index of the ``)`` that matches the ``(`` at ``start``.
2265 The walk skips single-quoted (``'...'``) and double-quoted (``"..."``)
2266 string literals, so parentheses inside string literals do not affect
2267 the depth counter. ``''`` and ``""`` are treated as escaped quotes
2268 inside their respective contexts, matching PostgreSQL/SQLite literal
2269 conventions.
2271 Returns ``None`` if the opening parenthesis is never closed
2272 (unbalanced). The character at ``text[start]`` must be ``(``.
2274 Note for SQLite use, SQLite also supports MySQL backtick-style quotes as
2275 well as SQL Server bracket style quotes; the latter has different escaping
2276 behaviors. A follow-up patch could add support for these two additional
2277 styles (consider using an enum like QuotingStyle.DOUBLE |
2278 QuotingStyle.BRACKET, etc.)
2280 E.g.::
2282 >>> find_matching_paren("(a + b)")
2283 6
2284 >>> find_matching_paren("((a)(b))")
2285 7
2286 >>> find_matching_paren("(a = '(' AND b = ')')")
2287 20
2289 """
2290 assert text[start] == "(", "start index must point at an open paren"
2292 depth = 0
2293 in_single = False
2294 in_double = False
2295 n = len(text)
2296 i = start
2297 while i < n:
2298 ch = text[i]
2299 if in_single:
2300 if ch == "'":
2301 if i + 1 < n and text[i + 1] == "'":
2302 i += 2
2303 continue
2304 in_single = False
2305 elif in_double:
2306 if ch == '"':
2307 if i + 1 < n and text[i + 1] == '"':
2308 i += 2
2309 continue
2310 in_double = False
2311 elif ch == "'":
2312 in_single = True
2313 elif ch == '"':
2314 in_double = True
2315 elif ch == "(":
2316 depth += 1
2317 elif ch == ")":
2318 depth -= 1
2319 if depth == 0:
2320 return i
2321 i += 1
2322 return None
2325def strip_outer_parens(text: str) -> str:
2326 """Remove one layer of outer parentheses from ``text`` if they wrap the
2327 entire (stripped) string.
2329 Whitespace is preserved if the parentheses do not wrap the whole
2330 expression. String literals are honored via :func:`find_matching_paren`,
2331 so ``"(a = '(' AND b = ')')"`` correctly strips to
2332 ``"a = '(' AND b = ')'"`` rather than being interpreted as two separate
2333 paren groups.
2335 E.g.::
2337 >>> strip_outer_parens("(a IS NOT NULL)")
2338 'a IS NOT NULL'
2339 >>> strip_outer_parens("(a) AND (b)")
2340 '(a) AND (b)'
2341 >>> strip_outer_parens("a NOT NULL")
2342 'a NOT NULL'
2344 """
2345 stripped = text.strip()
2346 lstripped = len(stripped)
2347 if lstripped < 2 or stripped[0] != "(" or stripped[-1] != ")":
2348 return text
2349 close = find_matching_paren(stripped, 0)
2350 if close is not None and close == lstripped - 1:
2351 return stripped[1:-1]
2352 return text
2355def quoted_token_parser(value):
2356 """Parse a dotted identifier with accommodation for quoted names.
2358 Includes support for SQL-style double quotes as a literal character.
2360 E.g.::
2362 >>> quoted_token_parser("name")
2363 ["name"]
2364 >>> quoted_token_parser("schema.name")
2365 ["schema", "name"]
2366 >>> quoted_token_parser('"Schema"."Name"')
2367 ['Schema', 'Name']
2368 >>> quoted_token_parser('"Schema"."Name""Foo"')
2369 ['Schema', 'Name""Foo']
2371 """
2373 if '"' not in value:
2374 return value.split(".")
2376 # 0 = outside of quotes
2377 # 1 = inside of quotes
2378 state = 0
2379 result: List[List[str]] = [[]]
2380 idx = 0
2381 lv = len(value)
2382 while idx < lv:
2383 char = value[idx]
2384 if char == '"':
2385 if state == 1 and idx < lv - 1 and value[idx + 1] == '"':
2386 result[-1].append('"')
2387 idx += 1
2388 else:
2389 state ^= 1
2390 elif char == "." and state == 0:
2391 result.append([])
2392 else:
2393 result[-1].append(char)
2394 idx += 1
2396 return ["".join(token) for token in result]
2399def add_parameter_text(params: Any, text: str) -> Callable[[_F], _F]:
2400 params = _collections.to_list(params)
2402 def decorate(fn):
2403 doc = fn.__doc__ is not None and fn.__doc__ or ""
2404 if doc:
2405 doc = inject_param_text(doc, {param: text for param in params})
2406 fn.__doc__ = doc
2407 return fn
2409 return decorate
2412def _dedent_docstring(text: str) -> str:
2413 split_text = text.split("\n", 1)
2414 if len(split_text) == 1:
2415 return text
2416 else:
2417 firstline, remaining = split_text
2418 if not firstline.startswith(" "):
2419 return firstline + "\n" + textwrap.dedent(remaining)
2420 else:
2421 return textwrap.dedent(text)
2424def inject_docstring_text(
2425 given_doctext: Optional[str], injecttext: str, pos: int
2426) -> str:
2427 doctext: str = _dedent_docstring(given_doctext or "")
2428 lines = doctext.split("\n")
2429 if len(lines) == 1:
2430 lines.append("")
2431 injectlines = textwrap.dedent(injecttext).split("\n")
2432 if injectlines[0]:
2433 injectlines.insert(0, "")
2435 blanks = [num for num, line in enumerate(lines) if not line.strip()]
2436 blanks.insert(0, 0)
2438 inject_pos = blanks[min(pos, len(blanks) - 1)]
2440 lines = lines[0:inject_pos] + injectlines + lines[inject_pos:]
2441 return "\n".join(lines)
2444_param_reg = re.compile(r"(\s+):param (.+?):")
2447def inject_param_text(doctext: str, inject_params: Dict[str, str]) -> str:
2448 doclines = collections.deque(doctext.splitlines())
2449 lines = []
2451 # TODO: this is not working for params like ":param case_sensitive=True:"
2453 to_inject = None
2454 while doclines:
2455 line = doclines.popleft()
2457 m = _param_reg.match(line)
2459 if to_inject is None:
2460 if m:
2461 param = m.group(2).lstrip("*")
2462 if param in inject_params:
2463 # default indent to that of :param: plus one
2464 indent = " " * len(m.group(1)) + " "
2466 # but if the next line has text, use that line's
2467 # indentation
2468 if doclines:
2469 m2 = re.match(r"(\s+)\S", doclines[0])
2470 if m2:
2471 indent = " " * len(m2.group(1))
2473 to_inject = indent + inject_params[param]
2474 elif m:
2475 lines.extend(["\n", to_inject, "\n"])
2476 to_inject = None
2477 elif not line.rstrip():
2478 lines.extend([line, to_inject, "\n"])
2479 to_inject = None
2480 elif line.endswith("::"):
2481 # TODO: this still won't cover if the code example itself has
2482 # blank lines in it, need to detect those via indentation.
2483 lines.extend([line, doclines.popleft()])
2484 continue
2485 lines.append(line)
2487 return "\n".join(lines)
2490def repr_tuple_names(names: List[str]) -> Optional[str]:
2491 """Trims a list of strings from the middle and return a string of up to
2492 four elements. Strings greater than 11 characters will be truncated"""
2493 if len(names) == 0:
2494 return None
2495 flag = len(names) <= 4
2496 names = names[0:4] if flag else names[0:3] + names[-1:]
2497 res = ["%s.." % name[:11] if len(name) > 11 else name for name in names]
2498 if flag:
2499 return ", ".join(res)
2500 else:
2501 return "%s, ..., %s" % (", ".join(res[0:3]), res[-1])
2504def has_compiled_ext(raise_=False):
2505 from ._has_cython import HAS_CYEXTENSION
2507 if HAS_CYEXTENSION:
2508 return True
2509 elif raise_:
2510 raise ImportError(
2511 "cython extensions were expected to be installed, "
2512 "but are not present"
2513 )
2514 else:
2515 return False
2518def load_uncompiled_module(module: _M) -> _M:
2519 """Load the non-compied version of a module that is also
2520 compiled with cython.
2521 """
2522 full_name = module.__name__
2523 assert module.__spec__
2524 parent_name = module.__spec__.parent
2525 assert parent_name
2526 parent_module = sys.modules[parent_name]
2527 assert parent_module.__spec__
2528 package_path = parent_module.__spec__.origin
2529 assert package_path and package_path.endswith("__init__.py")
2531 name = full_name.split(".")[-1]
2532 module_path = package_path.replace("__init__.py", f"{name}.py")
2534 py_spec = importlib.util.spec_from_file_location(full_name, module_path)
2535 assert py_spec
2536 py_module = importlib.util.module_from_spec(py_spec)
2537 assert py_spec.loader
2538 py_spec.loader.exec_module(py_module)
2539 return cast(_M, py_module)
2542_pre_release_normalize = {
2543 "a": "a",
2544 "alpha": "a",
2545 "b": "b",
2546 "beta": "b",
2547 "c": "rc",
2548 "pre": "rc",
2549 "preview": "rc",
2550 "rc": "rc",
2551}
2553_version_string_re = re.compile(
2554 r"""
2555 \s*
2556 (?:[a-z][a-z0-9]*[-_])? # ignored prefix, "py3-"
2557 v?
2558 (?P<release>\d+(?:\.\d+)*)
2559 (?: # pre-release
2560 [-_.]?
2561 (?P<pre_l>alpha|beta|preview|pre|rc|a|b|c)
2562 [-_.]?
2563 (?P<pre_n>\d+)?
2564 )?
2565 (?: # post-release
2566 [-_.]?
2567 (?P<post_l>post|rev|r)
2568 [-_.]?
2569 (?P<post_n>\d+)?
2570 )?
2571 (?: # developmental release
2572 [-_.]?
2573 (?P<dev_l>dev)
2574 [-_.]?
2575 (?P<dev_n>\d+)?
2576 )?
2577 """,
2578 re.X | re.I,
2579)
2581_VersionSortKey = Tuple[
2582 Tuple[int, ...],
2583 Tuple[int, str, int],
2584 Tuple[int, int],
2585 Tuple[int, int],
2586]
2589def _version_sort_key(
2590 release: Tuple[int, ...],
2591 pre: Optional[Tuple[str, int]],
2592 post: Optional[int],
2593 dev: Optional[int],
2594) -> _VersionSortKey:
2595 if pre is None and post is None and dev is not None:
2596 # a dev release with no other qualifiers precedes every
2597 # pre-release of the same release number
2598 pre_key = (-1, "", 0)
2599 elif pre is None:
2600 pre_key = (1, "", 0)
2601 else:
2602 pre_key = (0, pre[0], pre[1])
2604 return (
2605 release,
2606 pre_key,
2607 (0, 0) if post is None else (1, post),
2608 (1, 0) if dev is None else (0, dev),
2609 )
2612def _version_comparison(
2613 op: Callable[[Any, Any], bool],
2614) -> Callable[[VersionInfo, Any], Any]:
2615 """Build one of :class:`.VersionInfo`'s comparison methods.
2617 Comparison takes place against the sort key rather than the tuple
2618 itself, so that pre-release and similar qualifiers are taken into
2619 account. A plain tuple is interpreted as the release segment of a
2620 final release; anything else is not comparable.
2622 """
2624 def compare(self: VersionInfo, other: Any) -> Any:
2625 if isinstance(other, VersionInfo):
2626 other_key = other._sort_key
2627 elif isinstance(other, tuple):
2628 other_key = _version_sort_key(other, None, None, None)
2629 else:
2630 return NotImplemented
2631 return op(self._sort_key, other_key)
2633 return compare
2636class VersionInfo(Tuple[int, ...]):
2637 """A version number, as a tuple of integers.
2639 :class:`.VersionInfo` is a ``tuple`` subclass consisting of the
2640 numeric "release" segment of a version only, e.g. ``2.0.0rc1``
2641 is the tuple ``(2, 0, 0)``. Ordering however takes any
2642 pre-release, post-release and developmental qualifiers into account
2643 as described by :pep:`440`, so that ``2.0.0rc1`` compares as less than
2644 ``2.0.0``, including when compared against a plain tuple such as
2645 ``(2, 0, 0)``.
2647 Plain tuples are interpreted as final releases when compared against
2648 a :class:`.VersionInfo`.
2650 .. versionadded:: 2.1
2652 """
2654 string: Optional[str]
2655 """the string from which this version was parsed, if any."""
2657 pre: Optional[Tuple[str, int]]
2658 """normalized pre-release qualifier, e.g. ``("rc", 1)``."""
2660 post: Optional[int]
2661 """post-release number, if any."""
2663 dev: Optional[int]
2664 """developmental release number, if any."""
2666 _sort_key: _VersionSortKey
2668 def __new__(
2669 cls,
2670 release: Sequence[int] = (),
2671 *,
2672 string: Optional[str] = None,
2673 pre: Optional[Tuple[str, int]] = None,
2674 post: Optional[int] = None,
2675 dev: Optional[int] = None,
2676 ) -> VersionInfo:
2677 # __new__ is needed as the release segment has to be passed to
2678 # tuple.__new__(); the remaining state is set up in __init__
2679 return tuple.__new__(cls, release)
2681 def __init__(
2682 self,
2683 release: Sequence[int] = (),
2684 *,
2685 string: Optional[str] = None,
2686 pre: Optional[Tuple[str, int]] = None,
2687 post: Optional[int] = None,
2688 dev: Optional[int] = None,
2689 ):
2690 self.string = string
2691 self.pre = pre
2692 self.post = post
2693 self.dev = dev
2694 self._sort_key = _version_sort_key(tuple(self), pre, post, dev)
2696 def __repr__(self) -> str:
2697 if self.string is not None:
2698 return f"VersionInfo({tuple(self)!r}, string={self.string!r})"
2699 else:
2700 return f"VersionInfo({tuple(self)!r})"
2702 def __str__(self) -> str:
2703 if self.string is not None:
2704 return self.string
2705 else:
2706 return ".".join(str(num) for num in self)
2708 # every comparison has to be stated explicitly; ``tuple`` implements
2709 # all six of them, so ``functools.total_ordering`` fills in nothing
2710 # here and the ones left out would silently compare as plain tuples
2711 __eq__ = _version_comparison(operator.eq)
2712 __ne__ = _version_comparison(operator.ne)
2713 __lt__ = _version_comparison(operator.lt)
2714 __le__ = _version_comparison(operator.le)
2715 __gt__ = _version_comparison(operator.gt)
2716 __ge__ = _version_comparison(operator.ge)
2718 def __hash__(self) -> int:
2719 return hash(self._sort_key)
2722def parse_version_string(version: Optional[str]) -> VersionInfo:
2723 """Parse a DBAPI version string into a :class:`.VersionInfo`.
2725 Leading characters that are not part of the version itself are
2726 ignored, as are trailing characters following the version, so that
2727 strings such as ``"py3-4.0.19-beta4"`` and
2728 ``"2.9.10 (dt dec pq3 ext lo64)"`` parse correctly.
2730 An empty :class:`.VersionInfo` is returned if no version number can be
2731 located at all.
2733 Parsing is deliberately more tolerant than that of :pep:`440`, which
2734 the version strings published by DBAPIs frequently do not conform to;
2735 a strict implementation such as that of the ``packaging`` library
2736 rejects each of the above outright.
2738 .. versionadded:: 2.1
2740 """
2742 if not version:
2743 return VersionInfo((), string=version)
2745 m = _version_string_re.match(version)
2746 if m is None:
2747 return VersionInfo((), string=version)
2749 release = tuple(int(x) for x in m.group("release").split("."))
2751 pre_l = m.group("pre_l")
2752 pre: Optional[Tuple[str, int]]
2753 if pre_l is not None:
2754 pre = (
2755 _pre_release_normalize[pre_l.lower()],
2756 int(m.group("pre_n") or 0),
2757 )
2758 else:
2759 pre = None
2761 return VersionInfo(
2762 release,
2763 string=version,
2764 pre=pre,
2765 post=(
2766 int(m.group("post_n") or 0)
2767 if m.group("post_l") is not None
2768 else None
2769 ),
2770 dev=(
2771 int(m.group("dev_n") or 0)
2772 if m.group("dev_l") is not None
2773 else None
2774 ),
2775 )
2778def parse_version_from_metadata(distribution: str) -> VersionInfo:
2779 """Return the version of an installed distribution as a
2780 :class:`.VersionInfo`.
2782 This is intended for use by dialects whose DBAPI module does not
2783 itself publish a version number, such as ``asyncmy``. As the
2784 distribution name is not necessarily the same as the module name, and
2785 the installed distribution is not necessarily the module that was
2786 imported, this should not be used when the DBAPI module provides a
2787 version of its own.
2789 An empty :class:`.VersionInfo` is returned if the distribution is not
2790 installed.
2792 .. versionadded:: 2.1
2794 """
2796 try:
2797 version = importlib.metadata.version(distribution)
2798 except importlib.metadata.PackageNotFoundError:
2799 return VersionInfo()
2800 else:
2801 return parse_version_string(version)
2804class _Missing(enum.Enum):
2805 Missing = enum.auto()
2808Missing = _Missing.Missing
2809MissingOr = Union[_T, Literal[_Missing.Missing]]