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 operator
24import re
25import sys
26import textwrap
27import threading
28import types
29from types import CodeType
30from types import ModuleType
31from typing import Any
32from typing import Callable
33from typing import cast
34from typing import Dict
35from typing import FrozenSet
36from typing import Generic
37from typing import Iterator
38from typing import List
39from typing import Literal
40from typing import NoReturn
41from typing import Optional
42from typing import overload
43from typing import Sequence
44from typing import Set
45from typing import Tuple
46from typing import Type
47from typing import TYPE_CHECKING
48from typing import TypeVar
49from typing import Union
50import warnings
52from . import _collections
53from . import compat
54from .. import exc
56_T = TypeVar("_T")
57_T_co = TypeVar("_T_co", covariant=True)
58_F = TypeVar("_F", bound=Callable[..., Any])
59_MA = TypeVar("_MA", bound="HasMemoized.memoized_attribute[Any]")
60_M = TypeVar("_M", bound=ModuleType)
63def restore_annotations(
64 cls: type, new_annotations: dict[str, Any]
65) -> Callable[[], None]:
66 """apply alternate annotations to a class, with a callable to restore
67 the pristine state of the former.
68 This is used strictly to provide dataclasses on a mapped class, where
69 in some cases where are making dataclass fields based on an attribute
70 that is actually a python descriptor on a superclass which we called
71 to get a value.
72 if dataclasses were to give us a way to achieve this without swapping
73 __annotations__, that would be much better.
74 """
75 delattr_ = object()
77 # pep-649 means classes have "__annotate__", and it's a callable. if it's
78 # there and is None, we're in "legacy future mode", where it's python 3.14
79 # or higher and "from __future__ import annotations" is set. in "legacy
80 # future mode" we have to do the same steps we do for older pythons,
81 # __annotate__ can be ignored
82 is_pep649 = hasattr(cls, "__annotate__") and cls.__annotate__ is not None
84 if is_pep649:
85 memoized = {
86 "__annotate__": getattr(cls, "__annotate__", delattr_),
87 }
88 else:
89 memoized = {
90 "__annotations__": getattr(cls, "__annotations__", delattr_)
91 }
93 cls.__annotations__ = new_annotations
95 def restore():
96 for k, v in memoized.items():
97 if v is delattr_:
98 delattr(cls, k)
99 else:
100 setattr(cls, k, v)
102 return restore
105def md5_hex(x: Any) -> str:
106 x = x.encode("utf-8")
107 m = compat.md5_not_for_security()
108 m.update(x)
109 return cast(str, m.hexdigest())
112class safe_reraise:
113 """Reraise an exception after invoking some
114 handler code.
116 Stores the existing exception info before
117 invoking so that it is maintained across a potential
118 coroutine context switch.
120 e.g.::
122 try:
123 sess.commit()
124 except:
125 with safe_reraise():
126 sess.rollback()
128 TODO: we should at some point evaluate current behaviors in this regard
129 based on current greenlet, gevent/eventlet implementations in Python 3, and
130 also see the degree to which our own asyncio (based on greenlet also) is
131 impacted by this. .rollback() will cause IO / context switch to occur in
132 all these scenarios; what happens to the exception context from an
133 "except:" block if we don't explicitly store it? Original issue was #2703.
135 """
137 __slots__ = ("_exc_info",)
139 _exc_info: Union[
140 None,
141 Tuple[
142 Type[BaseException],
143 BaseException,
144 types.TracebackType,
145 ],
146 Tuple[None, None, None],
147 ]
149 def __enter__(self) -> None:
150 self._exc_info = sys.exc_info()
152 def __exit__(
153 self,
154 type_: Optional[Type[BaseException]],
155 value: Optional[BaseException],
156 traceback: Optional[types.TracebackType],
157 ) -> NoReturn:
158 assert self._exc_info is not None
159 # see #2703 for notes
160 if type_ is None:
161 exc_type, exc_value, exc_tb = self._exc_info
162 assert exc_value is not None
163 self._exc_info = None # remove potential circular references
164 raise exc_value.with_traceback(exc_tb)
165 else:
166 self._exc_info = None # remove potential circular references
167 assert value is not None
168 raise value.with_traceback(traceback)
171def walk_subclasses(cls: Type[_T]) -> Iterator[Type[_T]]:
172 seen: Set[Any] = set()
174 stack = [cls]
175 while stack:
176 cls = stack.pop()
177 if cls in seen:
178 continue
179 else:
180 seen.add(cls)
181 stack.extend(cls.__subclasses__())
182 yield cls
185def string_or_unprintable(element: Any) -> str:
186 if isinstance(element, str):
187 return element
188 else:
189 try:
190 return str(element)
191 except Exception:
192 return "unprintable element %r" % element
195def clsname_as_plain_name(
196 cls: Type[Any], use_name: Optional[str] = None
197) -> str:
198 name = use_name or cls.__name__
199 return " ".join(n.lower() for n in re.findall(r"([A-Z][a-z]+|SQL)", name))
202def method_is_overridden(
203 instance_or_cls: Union[Type[Any], object],
204 against_method: Callable[..., Any],
205) -> bool:
206 """Return True if the two class methods don't match."""
208 if not isinstance(instance_or_cls, type):
209 current_cls = instance_or_cls.__class__
210 else:
211 current_cls = instance_or_cls
213 method_name = against_method.__name__
215 current_method: types.MethodType = getattr(current_cls, method_name)
217 return current_method != against_method
220def decode_slice(slc: slice) -> Tuple[Any, ...]:
221 """decode a slice object as sent to __getitem__.
223 takes into account the 2.5 __index__() method, basically.
225 """
226 ret: List[Any] = []
227 for x in slc.start, slc.stop, slc.step:
228 if hasattr(x, "__index__"):
229 x = x.__index__()
230 ret.append(x)
231 return tuple(ret)
234def _unique_symbols(used: Sequence[str], *bases: str) -> Iterator[str]:
235 used_set = set(used)
236 for base in bases:
237 pool = itertools.chain(
238 (base,),
239 map(lambda i: base + str(i), range(1000)),
240 )
241 for sym in pool:
242 if sym not in used_set:
243 used_set.add(sym)
244 yield sym
245 break
246 else:
247 raise NameError("exhausted namespace for symbol base %s" % base)
250def map_bits(fn: Callable[[int], Any], n: int) -> Iterator[Any]:
251 """Call the given function given each nonzero bit from n."""
253 while n:
254 b = n & (~n + 1)
255 yield fn(b)
256 n ^= b
259_Fn = TypeVar("_Fn", bound="Callable[..., Any]")
261# this seems to be in flux in recent mypy versions
264def decorator(target: Callable[..., Any]) -> Callable[[_Fn], _Fn]:
265 """A signature-matching decorator factory."""
267 def decorate(fn: _Fn) -> _Fn:
268 if not inspect.isfunction(fn) and not inspect.ismethod(fn):
269 raise Exception("not a decoratable function")
271 # Python 3.14 defer creating __annotations__ until its used.
272 # We do not want to create __annotations__ now.
273 annofunc = getattr(fn, "__annotate__", None)
274 if annofunc is not None:
275 fn.__annotate__ = None # type: ignore[union-attr]
276 try:
277 spec = compat.inspect_getfullargspec(fn)
278 finally:
279 fn.__annotate__ = annofunc # type: ignore[union-attr]
280 else:
281 spec = compat.inspect_getfullargspec(fn)
283 # Do not generate code for annotations.
284 # update_wrapper() copies the annotation from fn to decorated.
285 # We use dummy defaults for code generation to avoid having
286 # copy of large globals for compiling.
287 # We copy __defaults__ and __kwdefaults__ from fn to decorated.
288 empty_defaults = (None,) * len(spec.defaults or ())
289 empty_kwdefaults = dict.fromkeys(spec.kwonlydefaults or ())
290 spec = spec._replace(
291 annotations={},
292 defaults=empty_defaults,
293 kwonlydefaults=empty_kwdefaults,
294 )
296 names = (
297 tuple(cast("Tuple[str, ...]", spec[0]))
298 + cast("Tuple[str, ...]", spec[1:3])
299 + (fn.__name__,)
300 )
301 targ_name, fn_name = _unique_symbols(names, "target", "fn")
303 metadata: Dict[str, Optional[str]] = dict(target=targ_name, fn=fn_name)
304 metadata.update(format_argspec_plus(spec, grouped=False))
305 metadata["name"] = fn.__name__
307 if inspect.iscoroutinefunction(fn):
308 metadata["prefix"] = "async "
309 metadata["target_prefix"] = "await "
310 else:
311 metadata["prefix"] = ""
312 metadata["target_prefix"] = ""
314 # look for __ positional arguments. This is a convention in
315 # SQLAlchemy that arguments should be passed positionally
316 # rather than as keyword
317 # arguments. note that apply_pos doesn't currently work in all cases
318 # such as when a kw-only indicator "*" is present, which is why
319 # we limit the use of this to just that case we can detect. As we add
320 # more kinds of methods that use @decorator, things may have to
321 # be further improved in this area
322 if "__" in repr(spec[0]):
323 code = """\
324%(prefix)sdef %(name)s%(grouped_args)s:
325 return %(target_prefix)s%(target)s(%(fn)s, %(apply_pos)s)
326""" % metadata
327 else:
328 code = """\
329%(prefix)sdef %(name)s%(grouped_args)s:
330 return %(target_prefix)s%(target)s(%(fn)s, %(apply_kw)s)
331""" % metadata
333 env: Dict[str, Any] = {
334 targ_name: target,
335 fn_name: fn,
336 "__name__": fn.__module__,
337 }
339 decorated = cast(
340 types.FunctionType,
341 _exec_code_in_env(code, env, fn.__name__),
342 )
343 decorated.__defaults__ = fn.__defaults__
344 decorated.__kwdefaults__ = fn.__kwdefaults__ # type: ignore[union-attr] # noqa: E501
345 return update_wrapper(decorated, fn) # type: ignore[return-value]
347 return update_wrapper(decorate, target) # type: ignore[return-value]
350def _exec_code_in_env(
351 code: Union[str, types.CodeType], env: Dict[str, Any], fn_name: str
352) -> Callable[..., Any]:
353 exec(code, env)
354 return env[fn_name] # type: ignore[no-any-return]
357_PF = TypeVar("_PF")
358_TE = TypeVar("_TE")
361class PluginLoader:
362 def __init__(
363 self, group: str, auto_fn: Optional[Callable[..., Any]] = None
364 ):
365 self.group = group
366 self.impls: Dict[str, Any] = {}
367 self.auto_fn = auto_fn
369 def clear(self):
370 self.impls.clear()
372 def load(self, name: str) -> Any:
373 if name in self.impls:
374 return self.impls[name]()
376 if self.auto_fn:
377 loader = self.auto_fn(name)
378 if loader:
379 self.impls[name] = loader
380 return loader()
382 for impl in compat.importlib_metadata_get(self.group):
383 if impl.name == name:
384 self.impls[name] = impl.load
385 return impl.load()
387 raise exc.NoSuchModuleError(
388 "Can't load plugin: %s:%s" % (self.group, name)
389 )
391 def register(self, name: str, modulepath: str, objname: str) -> None:
392 def load():
393 mod = __import__(modulepath)
394 for token in modulepath.split(".")[1:]:
395 mod = getattr(mod, token)
396 return getattr(mod, objname)
398 self.impls[name] = load
400 def deregister(self, name: str) -> None:
401 del self.impls[name]
404def _inspect_func_args(fn):
405 try:
406 co_varkeywords = inspect.CO_VARKEYWORDS
407 except AttributeError:
408 # https://docs.python.org/3/library/inspect.html
409 # The flags are specific to CPython, and may not be defined in other
410 # Python implementations. Furthermore, the flags are an implementation
411 # detail, and can be removed or deprecated in future Python releases.
412 spec = compat.inspect_getfullargspec(fn)
413 return spec[0], bool(spec[2])
414 else:
415 # use fn.__code__ plus flags to reduce method call overhead
416 co = fn.__code__
417 nargs = co.co_argcount
418 return (
419 list(co.co_varnames[:nargs]),
420 bool(co.co_flags & co_varkeywords),
421 )
424@overload
425def get_cls_kwargs(
426 cls: type,
427 *,
428 _set: Optional[Set[str]] = None,
429 raiseerr: Literal[True] = ...,
430) -> Set[str]: ...
433@overload
434def get_cls_kwargs(
435 cls: type, *, _set: Optional[Set[str]] = None, raiseerr: bool = False
436) -> Optional[Set[str]]: ...
439def get_cls_kwargs(
440 cls: type, *, _set: Optional[Set[str]] = None, raiseerr: bool = False
441) -> Optional[Set[str]]:
442 r"""Return the full set of inherited kwargs for the given `cls`.
444 Probes a class's __init__ method, collecting all named arguments. If the
445 __init__ defines a \**kwargs catch-all, then the constructor is presumed
446 to pass along unrecognized keywords to its base classes, and the
447 collection process is repeated recursively on each of the bases.
449 Uses a subset of inspect.getfullargspec() to cut down on method overhead,
450 as this is used within the Core typing system to create copies of type
451 objects which is a performance-sensitive operation.
453 No anonymous tuple arguments please !
455 """
456 toplevel = _set is None
457 if toplevel:
458 _set = set()
459 assert _set is not None
461 ctr = cls.__dict__.get("__init__", False)
463 has_init = (
464 ctr
465 and isinstance(ctr, types.FunctionType)
466 and isinstance(ctr.__code__, types.CodeType)
467 )
469 if has_init:
470 names, has_kw = _inspect_func_args(ctr)
471 _set.update(names)
473 if not has_kw and not toplevel:
474 if raiseerr:
475 raise TypeError(
476 f"given cls {cls} doesn't have an __init__ method"
477 )
478 else:
479 return None
480 else:
481 has_kw = False
483 if not has_init or has_kw:
484 for c in cls.__bases__:
485 if get_cls_kwargs(c, _set=_set) is None:
486 break
488 _set.discard("self")
489 return _set
492def get_func_kwargs(func: Callable[..., Any]) -> List[str]:
493 """Return the set of legal kwargs for the given `func`.
495 Uses getargspec so is safe to call for methods, functions,
496 etc.
498 """
500 return compat.inspect_getfullargspec(func)[0]
503def get_callable_argspec(
504 fn: Callable[..., Any], no_self: bool = False, _is_init: bool = False
505) -> compat.FullArgSpec:
506 """Return the argument signature for any callable.
508 All pure-Python callables are accepted, including
509 functions, methods, classes, objects with __call__;
510 builtins and other edge cases like functools.partial() objects
511 raise a TypeError.
513 """
514 if inspect.isbuiltin(fn):
515 raise TypeError("Can't inspect builtin: %s" % fn)
516 elif inspect.isfunction(fn) or (
517 hasattr(fn, "__code__")
518 and not inspect.isclass(fn)
519 and not inspect.ismethod(fn)
520 ):
521 if _is_init and no_self:
522 spec = compat.inspect_getfullargspec(fn)
523 return compat.FullArgSpec(
524 spec.args[1:],
525 spec.varargs,
526 spec.varkw,
527 spec.defaults,
528 spec.kwonlyargs,
529 spec.kwonlydefaults,
530 spec.annotations,
531 )
532 else:
533 return compat.inspect_getfullargspec(fn)
534 elif inspect.ismethod(fn):
535 if no_self and (_is_init or fn.__self__):
536 spec = compat.inspect_getfullargspec(fn.__func__)
537 return compat.FullArgSpec(
538 spec.args[1:],
539 spec.varargs,
540 spec.varkw,
541 spec.defaults,
542 spec.kwonlyargs,
543 spec.kwonlydefaults,
544 spec.annotations,
545 )
546 else:
547 return compat.inspect_getfullargspec(fn.__func__)
548 elif inspect.isclass(fn):
549 return get_callable_argspec(
550 fn.__init__, no_self=no_self, _is_init=True
551 )
552 elif hasattr(fn, "__func__"):
553 return compat.inspect_getfullargspec(fn.__func__)
554 elif hasattr(fn, "__call__"):
555 if inspect.ismethod(fn.__call__):
556 return get_callable_argspec(fn.__call__, no_self=no_self)
557 else:
558 raise TypeError("Can't inspect callable: %s" % fn)
559 else:
560 raise TypeError("Can't inspect callable: %s" % fn)
563def format_argspec_plus(
564 fn: Union[Callable[..., Any], compat.FullArgSpec], grouped: bool = True
565) -> Dict[str, Optional[str]]:
566 """Returns a dictionary of formatted, introspected function arguments.
568 A enhanced variant of inspect.formatargspec to support code generation.
570 fn
571 An inspectable callable or tuple of inspect getargspec() results.
572 grouped
573 Defaults to True; include (parens, around, argument) lists
575 Returns:
577 args
578 Full inspect.formatargspec for fn
579 self_arg
580 The name of the first positional argument, varargs[0], or None
581 if the function defines no positional arguments.
582 apply_pos
583 args, re-written in calling rather than receiving syntax. Arguments are
584 passed positionally.
585 apply_kw
586 Like apply_pos, except keyword-ish args are passed as keywords.
587 apply_pos_proxied
588 Like apply_pos but omits the self/cls argument
590 Example::
592 >>> format_argspec_plus(lambda self, a, b, c=3, **d: 123)
593 {'grouped_args': '(self, a, b, c=3, **d)',
594 'self_arg': 'self',
595 'apply_kw': '(self, a, b, c=c, **d)',
596 'apply_pos': '(self, a, b, c, **d)'}
598 """
599 if callable(fn):
600 spec = compat.inspect_getfullargspec(fn)
601 else:
602 spec = fn
604 args = compat.inspect_formatargspec(*spec)
606 apply_pos = compat.inspect_formatargspec(
607 spec[0], spec[1], spec[2], None, spec[4]
608 )
610 if spec[0]:
611 self_arg = spec[0][0]
613 apply_pos_proxied = compat.inspect_formatargspec(
614 spec[0][1:], spec[1], spec[2], None, spec[4]
615 )
617 elif spec[1]:
618 # I'm not sure what this is
619 self_arg = "%s[0]" % spec[1]
621 apply_pos_proxied = apply_pos
622 else:
623 self_arg = None
624 apply_pos_proxied = apply_pos
626 num_defaults = 0
627 if spec[3]:
628 num_defaults += len(cast(Tuple[Any], spec[3]))
629 if spec[4]:
630 num_defaults += len(spec[4])
632 name_args = spec[0] + spec[4]
634 defaulted_vals: Union[List[str], Tuple[()]]
636 if num_defaults:
637 defaulted_vals = name_args[0 - num_defaults :]
638 else:
639 defaulted_vals = ()
641 apply_kw = compat.inspect_formatargspec(
642 name_args,
643 spec[1],
644 spec[2],
645 defaulted_vals,
646 formatvalue=lambda x: "=" + str(x),
647 )
649 if spec[0]:
650 apply_kw_proxied = compat.inspect_formatargspec(
651 name_args[1:],
652 spec[1],
653 spec[2],
654 defaulted_vals,
655 formatvalue=lambda x: "=" + str(x),
656 )
657 else:
658 apply_kw_proxied = apply_kw
660 if grouped:
661 return dict(
662 grouped_args=args,
663 self_arg=self_arg,
664 apply_pos=apply_pos,
665 apply_kw=apply_kw,
666 apply_pos_proxied=apply_pos_proxied,
667 apply_kw_proxied=apply_kw_proxied,
668 )
669 else:
670 return dict(
671 grouped_args=args,
672 self_arg=self_arg,
673 apply_pos=apply_pos[1:-1],
674 apply_kw=apply_kw[1:-1],
675 apply_pos_proxied=apply_pos_proxied[1:-1],
676 apply_kw_proxied=apply_kw_proxied[1:-1],
677 )
680def format_argspec_init(method, grouped=True):
681 """format_argspec_plus with considerations for typical __init__ methods
683 Wraps format_argspec_plus with error handling strategies for typical
684 __init__ cases:
686 .. sourcecode:: text
688 object.__init__ -> (self)
689 other unreflectable (usually C) -> (self, *args, **kwargs)
691 """
692 if method is object.__init__:
693 grouped_args = "(self)"
694 args = "(self)" if grouped else "self"
695 proxied = "()" if grouped else ""
696 else:
697 try:
698 return format_argspec_plus(method, grouped=grouped)
699 except TypeError:
700 grouped_args = "(self, *args, **kwargs)"
701 args = grouped_args if grouped else "self, *args, **kwargs"
702 proxied = "(*args, **kwargs)" if grouped else "*args, **kwargs"
703 return dict(
704 self_arg="self",
705 grouped_args=grouped_args,
706 apply_pos=args,
707 apply_kw=args,
708 apply_pos_proxied=proxied,
709 apply_kw_proxied=proxied,
710 )
713def create_proxy_methods(
714 target_cls: Type[Any],
715 target_cls_sphinx_name: str,
716 proxy_cls_sphinx_name: str,
717 classmethods: Sequence[str] = (),
718 methods: Sequence[str] = (),
719 attributes: Sequence[str] = (),
720 use_intermediate_variable: Sequence[str] = (),
721) -> Callable[[_T], _T]:
722 """A class decorator indicating attributes should refer to a proxy
723 class.
725 This decorator is now a "marker" that does nothing at runtime. Instead,
726 it is consumed by the tools/generate_proxy_methods.py script to
727 statically generate proxy methods and attributes that are fully
728 recognized by typing tools such as mypy.
730 """
732 def decorate(cls):
733 return cls
735 return decorate
738def getargspec_init(method):
739 """inspect.getargspec with considerations for typical __init__ methods
741 Wraps inspect.getargspec with error handling for typical __init__ cases:
743 .. sourcecode:: text
745 object.__init__ -> (self)
746 other unreflectable (usually C) -> (self, *args, **kwargs)
748 """
749 try:
750 return compat.inspect_getfullargspec(method)
751 except TypeError:
752 if method is object.__init__:
753 return (["self"], None, None, None)
754 else:
755 return (["self"], "args", "kwargs", None)
758def unbound_method_to_callable(func_or_cls):
759 """Adjust the incoming callable such that a 'self' argument is not
760 required.
762 """
764 if isinstance(func_or_cls, types.MethodType) and not func_or_cls.__self__:
765 return func_or_cls.__func__
766 else:
767 return func_or_cls
770class GenericRepr:
771 """Encapsulates the logic for creating a generic __repr__() string.
773 This class allows for the repr structure to be created, then modified
774 (e.g., changing the class name), before being rendered as a string.
776 .. versionadded:: 2.1
777 """
779 __slots__ = (
780 "_obj",
781 "_additional_kw",
782 "_to_inspect",
783 "_omit_kwarg",
784 "_class_name",
785 )
787 _obj: Any
788 _additional_kw: Sequence[Tuple[str, Any]]
789 _to_inspect: List[object]
790 _omit_kwarg: Sequence[str]
791 _class_name: Optional[str]
793 def __init__(
794 self,
795 obj: Any,
796 additional_kw: Sequence[Tuple[str, Any]] = (),
797 to_inspect: Optional[Union[object, List[object]]] = None,
798 omit_kwarg: Sequence[str] = (),
799 ):
800 """Create a GenericRepr object.
802 :param obj: The object being repr'd
803 :param additional_kw: Additional keyword arguments to check for in
804 the repr, as a sequence of 2-tuples of (name, default_value)
805 :param to_inspect: One or more objects whose __init__ signature
806 should be inspected. If not provided, defaults to [obj].
807 :param omit_kwarg: Sequence of keyword argument names to omit from
808 the repr output
809 """
810 self._obj = obj
811 self._additional_kw = additional_kw
812 self._to_inspect = (
813 [obj] if to_inspect is None else _collections.to_list(to_inspect)
814 )
815 self._omit_kwarg = omit_kwarg
816 self._class_name = None
818 def set_class_name(self, class_name: str) -> GenericRepr:
819 """Set the class name to be used in the repr.
821 By default, the class name is taken from obj.__class__.__name__.
822 This method allows it to be overridden.
824 :param class_name: The class name to use
825 :return: self, for method chaining
826 """
827 self._class_name = class_name
828 return self
830 def __str__(self) -> str:
831 """Produce the __repr__() string based on the configured parameters."""
832 obj = self._obj
833 to_inspect = self._to_inspect
834 additional_kw = self._additional_kw
835 omit_kwarg = self._omit_kwarg
837 missing = object()
839 pos_args = []
840 kw_args: _collections.OrderedDict[str, Any] = (
841 _collections.OrderedDict()
842 )
843 vargs = None
844 for i, insp in enumerate(to_inspect):
845 try:
846 spec = compat.inspect_getfullargspec(insp.__init__) # type: ignore[misc] # noqa: E501
847 except TypeError:
848 continue
849 else:
850 default_len = len(spec.defaults) if spec.defaults else 0
851 if i == 0:
852 if spec.varargs:
853 vargs = spec.varargs
854 if default_len:
855 pos_args.extend(spec.args[1:-default_len])
856 else:
857 pos_args.extend(spec.args[1:])
858 else:
859 kw_args.update(
860 [(arg, missing) for arg in spec.args[1:-default_len]]
861 )
863 if default_len:
864 assert spec.defaults
865 kw_args.update(
866 [
867 (arg, default)
868 for arg, default in zip(
869 spec.args[-default_len:], spec.defaults
870 )
871 ]
872 )
873 output: List[str] = []
875 output.extend(repr(getattr(obj, arg, None)) for arg in pos_args)
877 if vargs is not None and hasattr(obj, vargs):
878 output.extend([repr(val) for val in getattr(obj, vargs)])
880 for arg, defval in kw_args.items():
881 if arg in omit_kwarg:
882 continue
883 try:
884 val = getattr(obj, arg, missing)
885 if val is not missing and val != defval:
886 output.append("%s=%r" % (arg, val))
887 except Exception:
888 pass
890 if additional_kw:
891 for arg, defval in additional_kw:
892 try:
893 val = getattr(obj, arg, missing)
894 if val is not missing and val != defval:
895 output.append("%s=%r" % (arg, val))
896 except Exception:
897 pass
899 class_name = (
900 self._class_name
901 if self._class_name is not None
902 else obj.__class__.__name__
903 )
904 return "%s(%s)" % (class_name, ", ".join(output))
907def generic_repr(
908 obj: Any,
909 additional_kw: Sequence[Tuple[str, Any]] = (),
910 to_inspect: Optional[Union[object, List[object]]] = None,
911 omit_kwarg: Sequence[str] = (),
912) -> str:
913 """Produce a __repr__() based on direct association of the __init__()
914 specification vs. same-named attributes present.
916 """
917 return str(
918 GenericRepr(
919 obj,
920 additional_kw=additional_kw,
921 to_inspect=to_inspect,
922 omit_kwarg=omit_kwarg,
923 )
924 )
927def class_hierarchy(cls):
928 """Return an unordered sequence of all classes related to cls.
930 Traverses diamond hierarchies.
932 Fibs slightly: subclasses of builtin types are not returned. Thus
933 class_hierarchy(class A(object)) returns (A, object), not A plus every
934 class systemwide that derives from object.
936 """
938 hier = {cls}
939 process = list(cls.__mro__)
940 while process:
941 c = process.pop()
942 bases = (_ for _ in c.__bases__ if _ not in hier)
944 for b in bases:
945 process.append(b)
946 hier.add(b)
948 if c.__module__ == "builtins" or not hasattr(c, "__subclasses__"):
949 continue
951 for s in [
952 _
953 for _ in (
954 c.__subclasses__()
955 if not issubclass(c, type)
956 else c.__subclasses__(c)
957 )
958 if _ not in hier
959 ]:
960 process.append(s)
961 hier.add(s)
962 return list(hier)
965def iterate_attributes(cls):
966 """iterate all the keys and attributes associated
967 with a class, without using getattr().
969 Does not use getattr() so that class-sensitive
970 descriptors (i.e. property.__get__()) are not called.
972 """
973 keys = dir(cls)
974 for key in keys:
975 for c in cls.__mro__:
976 if key in c.__dict__:
977 yield (key, c.__dict__[key])
978 break
981def monkeypatch_proxied_specials(
982 into_cls,
983 from_cls,
984 skip=None,
985 only=None,
986 name="self.proxy",
987 from_instance=None,
988):
989 """Automates delegation of __specials__ for a proxying type."""
991 if only:
992 dunders = only
993 else:
994 if skip is None:
995 skip = (
996 "__slots__",
997 "__del__",
998 "__getattribute__",
999 "__metaclass__",
1000 "__getstate__",
1001 "__setstate__",
1002 )
1003 dunders = [
1004 m
1005 for m in dir(from_cls)
1006 if (
1007 m.startswith("__")
1008 and m.endswith("__")
1009 and not hasattr(into_cls, m)
1010 and m not in skip
1011 )
1012 ]
1014 for method in dunders:
1015 try:
1016 maybe_fn = getattr(from_cls, method)
1017 if not hasattr(maybe_fn, "__call__"):
1018 continue
1019 maybe_fn = getattr(maybe_fn, "__func__", maybe_fn)
1020 fn = cast(types.FunctionType, maybe_fn)
1022 except AttributeError:
1023 continue
1024 try:
1025 spec = compat.inspect_getfullargspec(fn)
1026 fn_args = compat.inspect_formatargspec(spec[0])
1027 d_args = compat.inspect_formatargspec(spec[0][1:])
1028 except TypeError:
1029 fn_args = "(self, *args, **kw)"
1030 d_args = "(*args, **kw)"
1032 py = (
1033 "def %(method)s%(fn_args)s: "
1034 "return %(name)s.%(method)s%(d_args)s" % locals()
1035 )
1037 env: Dict[str, types.FunctionType] = (
1038 from_instance is not None and {name: from_instance} or {}
1039 )
1040 exec(py, env)
1041 try:
1042 env[method].__defaults__ = fn.__defaults__
1043 except AttributeError:
1044 pass
1045 setattr(into_cls, method, env[method])
1048def methods_equivalent(meth1, meth2):
1049 """Return True if the two methods are the same implementation."""
1051 return getattr(meth1, "__func__", meth1) is getattr(
1052 meth2, "__func__", meth2
1053 )
1056def as_interface(obj, cls=None, methods=None, required=None):
1057 """Ensure basic interface compliance for an instance or dict of callables.
1059 Checks that ``obj`` implements public methods of ``cls`` or has members
1060 listed in ``methods``. If ``required`` is not supplied, implementing at
1061 least one interface method is sufficient. Methods present on ``obj`` that
1062 are not in the interface are ignored.
1064 If ``obj`` is a dict and ``dict`` does not meet the interface
1065 requirements, the keys of the dictionary are inspected. Keys present in
1066 ``obj`` that are not in the interface will raise TypeErrors.
1068 Raises TypeError if ``obj`` does not meet the interface criteria.
1070 In all passing cases, an object with callable members is returned. In the
1071 simple case, ``obj`` is returned as-is; if dict processing kicks in then
1072 an anonymous class is returned.
1074 obj
1075 A type, instance, or dictionary of callables.
1076 cls
1077 Optional, a type. All public methods of cls are considered the
1078 interface. An ``obj`` instance of cls will always pass, ignoring
1079 ``required``..
1080 methods
1081 Optional, a sequence of method names to consider as the interface.
1082 required
1083 Optional, a sequence of mandatory implementations. If omitted, an
1084 ``obj`` that provides at least one interface method is considered
1085 sufficient. As a convenience, required may be a type, in which case
1086 all public methods of the type are required.
1088 """
1089 if not cls and not methods:
1090 raise TypeError("a class or collection of method names are required")
1092 if isinstance(cls, type) and isinstance(obj, cls):
1093 return obj
1095 interface = set(methods or [m for m in dir(cls) if not m.startswith("_")])
1096 implemented = set(dir(obj))
1098 complies = operator.ge
1099 if isinstance(required, type):
1100 required = interface
1101 elif not required:
1102 required = set()
1103 complies = operator.gt
1104 else:
1105 required = set(required)
1107 if complies(implemented.intersection(interface), required):
1108 return obj
1110 # No dict duck typing here.
1111 if not isinstance(obj, dict):
1112 qualifier = complies is operator.gt and "any of" or "all of"
1113 raise TypeError(
1114 "%r does not implement %s: %s"
1115 % (obj, qualifier, ", ".join(interface))
1116 )
1118 class AnonymousInterface:
1119 """A callable-holding shell."""
1121 if cls:
1122 AnonymousInterface.__name__ = "Anonymous" + cls.__name__
1123 found = set()
1125 for method, impl in dictlike_iteritems(obj):
1126 if method not in interface:
1127 raise TypeError("%r: unknown in this interface" % method)
1128 if not callable(impl):
1129 raise TypeError("%r=%r is not callable" % (method, impl))
1130 setattr(AnonymousInterface, method, staticmethod(impl))
1131 found.add(method)
1133 if complies(found, required):
1134 return AnonymousInterface
1136 raise TypeError(
1137 "dictionary does not contain required keys %s"
1138 % ", ".join(required - found)
1139 )
1142_GFD = TypeVar("_GFD", bound="generic_fn_descriptor[Any]")
1145class generic_fn_descriptor(Generic[_T_co]):
1146 """Descriptor which proxies a function when the attribute is not
1147 present in dict
1149 This superclass is organized in a particular way with "memoized" and
1150 "non-memoized" implementation classes that are hidden from type checkers,
1151 as Mypy seems to not be able to handle seeing multiple kinds of descriptor
1152 classes used for the same attribute.
1154 """
1156 fget: Callable[..., _T_co]
1157 __doc__: Optional[str]
1158 __name__: str
1160 def __init__(self, fget: Callable[..., _T_co], doc: Optional[str] = None):
1161 self.fget = fget
1162 self.__doc__ = doc or fget.__doc__
1163 self.__name__ = fget.__name__
1165 @overload
1166 def __get__(self: _GFD, obj: None, cls: Any) -> _GFD: ...
1168 @overload
1169 def __get__(self, obj: object, cls: Any) -> _T_co: ...
1171 def __get__(self: _GFD, obj: Any, cls: Any) -> Union[_GFD, _T_co]:
1172 raise NotImplementedError()
1174 if TYPE_CHECKING:
1176 def __set__(self, instance: Any, value: Any) -> None: ...
1178 def __delete__(self, instance: Any) -> None: ...
1180 def _reset(self, obj: Any) -> None:
1181 raise NotImplementedError()
1183 @classmethod
1184 def reset(cls, obj: Any, name: str) -> None:
1185 raise NotImplementedError()
1188class _non_memoized_property(generic_fn_descriptor[_T_co]):
1189 """a plain descriptor that proxies a function.
1191 primary rationale is to provide a plain attribute that's
1192 compatible with memoized_property which is also recognized as equivalent
1193 by mypy.
1195 """
1197 if not TYPE_CHECKING:
1199 def __get__(self, obj, cls):
1200 if obj is None:
1201 return self
1202 return self.fget(obj)
1205class _memoized_property(generic_fn_descriptor[_T_co]):
1206 """A read-only @property that is only evaluated once."""
1208 if not TYPE_CHECKING:
1210 def __get__(self, obj, cls):
1211 if obj is None:
1212 return self
1213 obj.__dict__[self.__name__] = result = self.fget(obj)
1214 return result
1216 def _reset(self, obj):
1217 _memoized_property.reset(obj, self.__name__)
1219 @classmethod
1220 def reset(cls, obj, name):
1221 obj.__dict__.pop(name, None)
1224# despite many attempts to get Mypy to recognize an overridden descriptor
1225# where one is memoized and the other isn't, there seems to be no reliable
1226# way other than completely deceiving the type checker into thinking there
1227# is just one single descriptor type everywhere. Otherwise, if a superclass
1228# has non-memoized and subclass has memoized, that requires
1229# "class memoized(non_memoized)". but then if a superclass has memoized and
1230# superclass has non-memoized, the class hierarchy of the descriptors
1231# would need to be reversed; "class non_memoized(memoized)". so there's no
1232# way to achieve this.
1233# additional issues, RO properties:
1234# https://github.com/python/mypy/issues/12440
1235if TYPE_CHECKING:
1236 # allow memoized and non-memoized to be freely mixed by having them
1237 # be the same class
1238 memoized_property = generic_fn_descriptor
1239 non_memoized_property = generic_fn_descriptor
1241 # for read only situations, mypy only sees @property as read only.
1242 # read only is needed when a subtype specializes the return type
1243 # of a property, meaning assignment needs to be disallowed
1244 ro_memoized_property = property
1245 ro_non_memoized_property = property
1247else:
1248 memoized_property = ro_memoized_property = _memoized_property
1249 non_memoized_property = ro_non_memoized_property = _non_memoized_property
1252def memoized_instancemethod(fn: _F) -> _F:
1253 """Decorate a method memoize its return value.
1255 Best applied to no-arg methods: memoization is not sensitive to
1256 argument values, and will always return the same value even when
1257 called with different arguments.
1259 """
1261 def oneshot(self, *args, **kw):
1262 result = fn(self, *args, **kw)
1264 def memo(*a, **kw):
1265 return result
1267 memo.__name__ = fn.__name__
1268 memo.__doc__ = fn.__doc__
1269 self.__dict__[fn.__name__] = memo
1270 return result
1272 return update_wrapper(oneshot, fn) # type: ignore[return-value]
1275class HasMemoized:
1276 """A mixin class that maintains the names of memoized elements in a
1277 collection for easy cache clearing, generative, etc.
1279 """
1281 if not TYPE_CHECKING:
1282 # support classes that want to have __slots__ with an explicit
1283 # slot for __dict__. not sure if that requires base __slots__ here.
1284 __slots__ = ()
1286 _memoized_keys: FrozenSet[str] = frozenset()
1288 def _reset_memoizations(self) -> None:
1289 for elem in self._memoized_keys:
1290 self.__dict__.pop(elem, None)
1292 def _assert_no_memoizations(self) -> None:
1293 for elem in self._memoized_keys:
1294 assert elem not in self.__dict__
1296 def _set_memoized_attribute(self, key: str, value: Any) -> None:
1297 self.__dict__[key] = value
1298 self._memoized_keys |= {key}
1300 class memoized_attribute(memoized_property[_T]):
1301 """A read-only @property that is only evaluated once.
1303 :meta private:
1305 """
1307 fget: Callable[..., _T]
1308 __doc__: Optional[str]
1309 __name__: str
1311 def __init__(self, fget: Callable[..., _T], doc: Optional[str] = None):
1312 self.fget = fget
1313 self.__doc__ = doc or fget.__doc__
1314 self.__name__ = fget.__name__
1316 @overload
1317 def __get__(self: _MA, obj: None, cls: Any) -> _MA: ...
1319 @overload
1320 def __get__(self, obj: Any, cls: Any) -> _T: ...
1322 def __get__(self, obj, cls):
1323 if obj is None:
1324 return self
1325 obj.__dict__[self.__name__] = result = self.fget(obj)
1326 obj._memoized_keys |= {self.__name__}
1327 return result
1329 @classmethod
1330 def memoized_instancemethod(cls, fn: _F) -> _F:
1331 """Decorate a method memoize its return value.
1333 :meta private:
1335 """
1337 def oneshot(self: Any, *args: Any, **kw: Any) -> Any:
1338 result = fn(self, *args, **kw)
1340 def memo(*a, **kw):
1341 return result
1343 memo.__name__ = fn.__name__
1344 memo.__doc__ = fn.__doc__
1345 self.__dict__[fn.__name__] = memo
1346 self._memoized_keys |= {fn.__name__}
1347 return result
1349 return update_wrapper(oneshot, fn) # type: ignore[return-value]
1352if TYPE_CHECKING:
1353 HasMemoized_ro_memoized_attribute = property
1354else:
1355 HasMemoized_ro_memoized_attribute = HasMemoized.memoized_attribute
1358class MemoizedSlots:
1359 """Apply memoized items to an object using a __getattr__ scheme.
1361 This allows the functionality of memoized_property and
1362 memoized_instancemethod to be available to a class using __slots__.
1364 The memoized get is not threadsafe under freethreading and the
1365 creator method may in extremely rare cases be called more than once.
1367 """
1369 __slots__ = ()
1371 def _fallback_getattr(self, key):
1372 raise AttributeError(key)
1374 def __getattr__(self, key: str) -> Any:
1375 if key.startswith("_memoized_attr_") or key.startswith(
1376 "_memoized_method_"
1377 ):
1378 raise AttributeError(key)
1379 # to avoid recursion errors when interacting with other __getattr__
1380 # schemes that refer to this one, when testing for memoized method
1381 # look at __class__ only rather than going into __getattr__ again.
1382 elif hasattr(self.__class__, f"_memoized_attr_{key}"):
1383 value = getattr(self, f"_memoized_attr_{key}")()
1384 setattr(self, key, value)
1385 return value
1386 elif hasattr(self.__class__, f"_memoized_method_{key}"):
1387 meth = getattr(self, f"_memoized_method_{key}")
1389 def oneshot(*args, **kw):
1390 result = meth(*args, **kw)
1392 def memo(*a, **kw):
1393 return result
1395 memo.__name__ = meth.__name__
1396 memo.__doc__ = meth.__doc__
1397 setattr(self, key, memo)
1398 return result
1400 oneshot.__doc__ = meth.__doc__
1401 return oneshot
1402 else:
1403 return self._fallback_getattr(key)
1406# from paste.deploy.converters
1407def asbool(obj: Any) -> bool:
1408 if isinstance(obj, str):
1409 obj = obj.strip().lower()
1410 if obj in ["true", "yes", "on", "y", "t", "1"]:
1411 return True
1412 elif obj in ["false", "no", "off", "n", "f", "0"]:
1413 return False
1414 else:
1415 raise ValueError("String is not true/false: %r" % obj)
1416 return bool(obj)
1419def bool_or_str(*text: str) -> Callable[[str], Union[str, bool]]:
1420 """Return a callable that will evaluate a string as
1421 boolean, or one of a set of "alternate" string values.
1423 """
1425 def bool_or_value(obj: str) -> Union[str, bool]:
1426 if obj in text:
1427 return obj
1428 else:
1429 return asbool(obj)
1431 return bool_or_value
1434def asint(value: Any) -> Optional[int]:
1435 """Coerce to integer."""
1437 if value is None:
1438 return value
1439 return int(value)
1442def coerce_kw_type(
1443 kw: Dict[str, Any],
1444 key: str,
1445 type_: Type[Any],
1446 flexi_bool: bool = True,
1447 dest: Optional[Dict[str, Any]] = None,
1448) -> None:
1449 r"""If 'key' is present in dict 'kw', coerce its value to type 'type\_' if
1450 necessary. If 'flexi_bool' is True, the string '0' is considered false
1451 when coercing to boolean.
1452 """
1454 if dest is None:
1455 dest = kw
1457 if (
1458 key in kw
1459 and (not isinstance(type_, type) or not isinstance(kw[key], type_))
1460 and kw[key] is not None
1461 ):
1462 if type_ is bool and flexi_bool:
1463 dest[key] = asbool(kw[key])
1464 else:
1465 dest[key] = type_(kw[key])
1468def constructor_key(obj: Any, cls: Type[Any]) -> Tuple[Any, ...]:
1469 """Produce a tuple structure that is cacheable using the __dict__ of
1470 obj to retrieve values
1472 """
1473 names = get_cls_kwargs(cls)
1474 return (cls,) + tuple(
1475 (k, obj.__dict__[k]) for k in names if k in obj.__dict__
1476 )
1479def constructor_copy(obj: _T, cls: Type[_T], *args: Any, **kw: Any) -> _T:
1480 """Instantiate cls using the __dict__ of obj as constructor arguments.
1482 Uses inspect to match the named arguments of ``cls``.
1484 """
1486 names = get_cls_kwargs(cls)
1487 kw.update(
1488 (k, obj.__dict__[k]) for k in names.difference(kw) if k in obj.__dict__
1489 )
1490 return cls(*args, **kw)
1493def counter() -> Callable[[], int]:
1494 """Return a threadsafe counter function."""
1496 lock = threading.Lock()
1497 counter = itertools.count(1)
1499 # avoid the 2to3 "next" transformation...
1500 def _next():
1501 with lock:
1502 return next(counter)
1504 return _next
1507def duck_type_collection(
1508 specimen: Any, default: Optional[Type[Any]] = None
1509) -> Optional[Type[Any]]:
1510 """Given an instance or class, guess if it is or is acting as one of
1511 the basic collection types: list, set and dict. If the __emulates__
1512 property is present, return that preferentially.
1513 """
1515 if hasattr(specimen, "__emulates__"):
1516 # canonicalize set vs sets.Set to a standard: the builtin set
1517 if specimen.__emulates__ is not None and issubclass(
1518 specimen.__emulates__, set
1519 ):
1520 return set
1521 else:
1522 return specimen.__emulates__ # type: ignore[no-any-return]
1524 isa = issubclass if isinstance(specimen, type) else isinstance
1525 if isa(specimen, list):
1526 return list
1527 elif isa(specimen, set):
1528 return set
1529 elif isa(specimen, dict):
1530 return dict
1532 if hasattr(specimen, "append"):
1533 return list
1534 elif hasattr(specimen, "add"):
1535 return set
1536 elif hasattr(specimen, "set"):
1537 return dict
1538 else:
1539 return default
1542def assert_arg_type(
1543 arg: Any, argtype: Union[Tuple[Type[Any], ...], Type[Any]], name: str
1544) -> Any:
1545 if isinstance(arg, argtype):
1546 return arg
1547 else:
1548 if isinstance(argtype, tuple):
1549 raise exc.ArgumentError(
1550 "Argument '%s' is expected to be one of type %s, got '%s'"
1551 % (name, " or ".join("'%s'" % a for a in argtype), type(arg))
1552 )
1553 else:
1554 raise exc.ArgumentError(
1555 "Argument '%s' is expected to be of type '%s', got '%s'"
1556 % (name, argtype, type(arg))
1557 )
1560def dictlike_iteritems(dictlike):
1561 """Return a (key, value) iterator for almost any dict-like object."""
1563 if hasattr(dictlike, "items"):
1564 return list(dictlike.items())
1566 getter = getattr(dictlike, "__getitem__", getattr(dictlike, "get", None))
1567 if getter is None:
1568 raise TypeError("Object '%r' is not dict-like" % dictlike)
1570 if hasattr(dictlike, "iterkeys"):
1572 def iterator():
1573 for key in dictlike.iterkeys():
1574 assert getter is not None
1575 yield key, getter(key)
1577 return iterator()
1578 elif hasattr(dictlike, "keys"):
1579 return iter((key, getter(key)) for key in dictlike.keys())
1580 else:
1581 raise TypeError("Object '%r' is not dict-like" % dictlike)
1584class classproperty(property):
1585 """A decorator that behaves like @property except that operates
1586 on classes rather than instances.
1588 The decorator is currently special when using the declarative
1589 module, but note that the
1590 :class:`~.sqlalchemy.ext.declarative.declared_attr`
1591 decorator should be used for this purpose with declarative.
1593 """
1595 fget: Callable[[Any], Any]
1597 def __init__(self, fget: Callable[[Any], Any], *arg: Any, **kw: Any):
1598 super().__init__(fget, *arg, **kw)
1599 self.__doc__ = fget.__doc__
1601 def __get__(self, obj: Any, cls: Optional[type] = None) -> Any:
1602 return self.fget(cls)
1605class hybridproperty(Generic[_T]):
1606 def __init__(self, func: Callable[..., _T]):
1607 self.func = func
1608 self.clslevel = func
1610 def __get__(self, instance: Any, owner: Any) -> _T:
1611 if instance is None:
1612 clsval = self.clslevel(owner)
1613 return clsval
1614 else:
1615 return self.func(instance)
1617 def classlevel(self, func: Callable[..., Any]) -> hybridproperty[_T]:
1618 self.clslevel = func
1619 return self
1622class rw_hybridproperty(Generic[_T]):
1623 def __init__(self, func: Callable[..., _T]):
1624 self.func = func
1625 self.clslevel = func
1626 self.setfn: Optional[Callable[..., Any]] = None
1628 def __get__(self, instance: Any, owner: Any) -> _T:
1629 if instance is None:
1630 clsval = self.clslevel(owner)
1631 return clsval
1632 else:
1633 return self.func(instance)
1635 def __set__(self, instance: Any, value: Any) -> None:
1636 assert self.setfn is not None
1637 self.setfn(instance, value)
1639 def setter(self, func: Callable[..., Any]) -> rw_hybridproperty[_T]:
1640 self.setfn = func
1641 return self
1643 def classlevel(self, func: Callable[..., Any]) -> rw_hybridproperty[_T]:
1644 self.clslevel = func
1645 return self
1648class hybridmethod(Generic[_T]):
1649 """Decorate a function as cls- or instance- level."""
1651 def __init__(self, func: Callable[..., _T]):
1652 self.func = self.__func__ = func
1653 self.clslevel = func
1655 def __get__(self, instance: Any, owner: Any) -> Callable[..., _T]:
1656 if instance is None:
1657 return self.clslevel.__get__( # type: ignore[no-any-return]
1658 owner, owner.__class__
1659 )
1660 else:
1661 return self.func.__get__( # type: ignore[no-any-return]
1662 instance, owner
1663 )
1665 def classlevel(self, func: Callable[..., Any]) -> hybridmethod[_T]:
1666 self.clslevel = func
1667 return self
1670class symbol(int):
1671 """A constant symbol.
1673 >>> symbol("foo") is symbol("foo")
1674 True
1675 >>> symbol("foo")
1676 <symbol 'foo>
1678 A slight refinement of the MAGICCOOKIE=object() pattern. The primary
1679 advantage of symbol() is its repr(). They are also singletons.
1681 Repeated calls of symbol('name') will all return the same instance.
1683 """
1685 name: str
1687 symbols: Dict[str, symbol] = {}
1688 _lock = threading.Lock()
1690 def __new__(
1691 cls,
1692 name: str,
1693 doc: Optional[str] = None,
1694 canonical: Optional[int] = None,
1695 ) -> symbol:
1696 with cls._lock:
1697 sym = cls.symbols.get(name)
1698 if sym is None:
1699 assert isinstance(name, str)
1700 if canonical is None:
1701 canonical = hash(name)
1702 sym = int.__new__(symbol, canonical)
1703 sym.name = name
1704 if doc:
1705 sym.__doc__ = doc
1707 # NOTE: we should ultimately get rid of this global thing,
1708 # however, currently it is to support pickling. The best
1709 # change would be when we are on py3.11 at a minimum, we
1710 # switch to stdlib enum.IntFlag.
1711 cls.symbols[name] = sym
1712 else:
1713 if canonical and canonical != sym:
1714 raise TypeError(
1715 f"Can't replace canonical symbol for {name!r} "
1716 f"with new int value {canonical}"
1717 )
1718 return sym
1720 def __reduce__(self):
1721 return symbol, (self.name, "x", int(self))
1723 def __str__(self):
1724 return repr(self)
1726 def __repr__(self):
1727 return f"symbol({self.name!r})"
1730class _IntFlagMeta(type):
1731 def __init__(
1732 cls,
1733 classname: str,
1734 bases: Tuple[Type[Any], ...],
1735 dict_: Dict[str, Any],
1736 **kw: Any,
1737 ) -> None:
1738 items: List[symbol]
1739 cls._items = items = []
1740 for k, v in dict_.items():
1741 if re.match(r"^__.*__$", k):
1742 continue
1743 if isinstance(v, int):
1744 sym = symbol(k, canonical=v)
1745 elif not k.startswith("_"):
1746 raise TypeError("Expected integer values for IntFlag")
1747 else:
1748 continue
1749 setattr(cls, k, sym)
1750 items.append(sym)
1752 cls.__members__ = _collections.immutabledict(
1753 {sym.name: sym for sym in items}
1754 )
1756 def __iter__(self) -> Iterator[symbol]:
1757 raise NotImplementedError(
1758 "iter not implemented to ensure compatibility with "
1759 "Python 3.11 IntFlag. Please use __members__. See "
1760 "https://github.com/python/cpython/issues/99304"
1761 )
1764class _FastIntFlag(metaclass=_IntFlagMeta):
1765 """An 'IntFlag' copycat that isn't slow when performing bitwise
1766 operations.
1768 the ``FastIntFlag`` class will return ``enum.IntFlag`` under TYPE_CHECKING
1769 and ``_FastIntFlag`` otherwise.
1771 """
1774if TYPE_CHECKING:
1775 from enum import IntFlag
1777 FastIntFlag = IntFlag
1778else:
1779 FastIntFlag = _FastIntFlag
1782_E = TypeVar("_E", bound=enum.Enum)
1785def parse_user_argument_for_enum(
1786 arg: Any,
1787 choices: Dict[_E, List[Any]],
1788 name: str,
1789 resolve_symbol_names: bool = False,
1790) -> Optional[_E]:
1791 """Given a user parameter, parse the parameter into a chosen value
1792 from a list of choice objects, typically Enum values.
1794 The user argument can be a string name that matches the name of a
1795 symbol, or the symbol object itself, or any number of alternate choices
1796 such as True/False/ None etc.
1798 :param arg: the user argument.
1799 :param choices: dictionary of enum values to lists of possible
1800 entries for each.
1801 :param name: name of the argument. Used in an :class:`.ArgumentError`
1802 that is raised if the parameter doesn't match any available argument.
1804 """
1805 for enum_value, choice in choices.items():
1806 if arg is enum_value:
1807 return enum_value
1808 elif resolve_symbol_names and arg == enum_value.name:
1809 return enum_value
1810 elif arg in choice:
1811 return enum_value
1813 if arg is None:
1814 return None
1816 raise exc.ArgumentError(f"Invalid value for '{name}': {arg!r}")
1819_creation_order = 1
1822def set_creation_order(instance: Any) -> None:
1823 """Assign a '_creation_order' sequence to the given instance.
1825 This allows multiple instances to be sorted in order of creation
1826 (typically within a single thread; the counter is not particularly
1827 threadsafe).
1829 """
1830 global _creation_order
1831 instance._creation_order = _creation_order
1832 _creation_order += 1
1835def warn_exception(func: Callable[..., Any], *args: Any, **kwargs: Any) -> Any:
1836 """executes the given function, catches all exceptions and converts to
1837 a warning.
1839 """
1840 try:
1841 return func(*args, **kwargs)
1842 except Exception:
1843 warn("%s('%s') ignored" % sys.exc_info()[0:2])
1846def ellipses_string(value, len_=25):
1847 try:
1848 if len(value) > len_:
1849 return "%s..." % value[0:len_]
1850 else:
1851 return value
1852 except TypeError:
1853 return value
1856class _hash_limit_string(str):
1857 """A string subclass that can only be hashed on a maximum amount
1858 of unique values.
1860 This is used for warnings so that we can send out parameterized warnings
1861 without the __warningregistry__ of the module, or the non-overridable
1862 "once" registry within warnings.py, overloading memory,
1865 """
1867 _hash: int
1869 def __new__(
1870 cls, value: str, num: int, args: Sequence[Any]
1871 ) -> _hash_limit_string:
1872 interpolated = (value % args) + (
1873 " (this warning may be suppressed after %d occurrences)" % num
1874 )
1875 self = super().__new__(cls, interpolated)
1876 self._hash = hash("%s_%d" % (value, hash(interpolated) % num))
1877 return self
1879 def __hash__(self) -> int:
1880 return self._hash
1882 def __eq__(self, other: Any) -> bool:
1883 return hash(self) == hash(other)
1886def warn(msg: str, code: Optional[str] = None) -> None:
1887 """Issue a warning.
1889 If msg is a string, :class:`.exc.SAWarning` is used as
1890 the category.
1892 """
1893 if code:
1894 _warnings_warn(exc.SAWarning(msg, code=code))
1895 else:
1896 _warnings_warn(msg, exc.SAWarning)
1899def warn_limited(msg: str, args: Sequence[Any]) -> None:
1900 """Issue a warning with a parameterized string, limiting the number
1901 of registrations.
1903 """
1904 if args:
1905 msg = _hash_limit_string(msg, 10, args)
1906 _warnings_warn(msg, exc.SAWarning)
1909_warning_tags: Dict[CodeType, Tuple[str, Type[Warning]]] = {}
1912def tag_method_for_warnings(
1913 message: str, category: Type[Warning]
1914) -> Callable[[_F], _F]:
1915 def go(fn):
1916 _warning_tags[fn.__code__] = (message, category)
1917 return fn
1919 return go
1922_not_sa_pattern = re.compile(r"^(?:sqlalchemy\.(?!testing)|alembic\.)")
1925def _warnings_warn(
1926 message: Union[str, Warning],
1927 category: Optional[Type[Warning]] = None,
1928 stacklevel: int = 2,
1929) -> None:
1931 if category is None and isinstance(message, Warning):
1932 category = type(message)
1934 # adjust the given stacklevel to be outside of SQLAlchemy
1935 try:
1936 frame = sys._getframe(stacklevel)
1937 except ValueError:
1938 # being called from less than 3 (or given) stacklevels, weird,
1939 # but don't crash
1940 stacklevel = 0
1941 except:
1942 # _getframe() doesn't work, weird interpreter issue, weird,
1943 # ok, but don't crash
1944 stacklevel = 0
1945 else:
1946 stacklevel_found = warning_tag_found = False
1947 while frame is not None:
1948 # using __name__ here requires that we have __name__ in the
1949 # __globals__ of the decorated string functions we make also.
1950 # we generate this using {"__name__": fn.__module__}
1951 if not stacklevel_found and not re.match(
1952 _not_sa_pattern, frame.f_globals.get("__name__", "")
1953 ):
1954 # stop incrementing stack level if an out-of-SQLA line
1955 # were found.
1956 stacklevel_found = True
1958 # however, for the warning tag thing, we have to keep
1959 # scanning up the whole traceback
1961 if frame.f_code in _warning_tags:
1962 warning_tag_found = True
1963 _suffix, _category = _warning_tags[frame.f_code]
1964 category = category or _category
1965 message = f"{message} ({_suffix})"
1967 frame = frame.f_back # type: ignore[assignment]
1969 if not stacklevel_found:
1970 stacklevel += 1
1971 elif stacklevel_found and warning_tag_found:
1972 break
1974 if category is not None:
1975 warnings.warn(message, category, stacklevel=stacklevel + 1)
1976 else:
1977 warnings.warn(message, stacklevel=stacklevel + 1)
1980def only_once(
1981 fn: Callable[..., _T], retry_on_exception: bool
1982) -> Callable[..., Optional[_T]]:
1983 """Decorate the given function to be a no-op after it is called exactly
1984 once."""
1986 once = [fn]
1988 def go(*arg: Any, **kw: Any) -> Optional[_T]:
1989 # strong reference fn so that it isn't garbage collected,
1990 # which interferes with the event system's expectations
1991 strong_fn = fn # noqa
1992 if once:
1993 once_fn = once.pop()
1994 try:
1995 return once_fn(*arg, **kw)
1996 except:
1997 if retry_on_exception:
1998 once.insert(0, once_fn)
1999 raise
2001 return None
2003 return go
2006_SQLA_RE = re.compile(r"sqlalchemy/([a-z_]+/){0,2}[a-z_]+\.py")
2007_UNITTEST_RE = re.compile(r"unit(?:2|test2?/)")
2010def chop_traceback(
2011 tb: List[str],
2012 exclude_prefix: re.Pattern[str] = _UNITTEST_RE,
2013 exclude_suffix: re.Pattern[str] = _SQLA_RE,
2014) -> List[str]:
2015 """Chop extraneous lines off beginning and end of a traceback.
2017 :param tb:
2018 a list of traceback lines as returned by ``traceback.format_stack()``
2020 :param exclude_prefix:
2021 a regular expression object matching lines to skip at beginning of
2022 ``tb``
2024 :param exclude_suffix:
2025 a regular expression object matching lines to skip at end of ``tb``
2026 """
2027 start = 0
2028 end = len(tb) - 1
2029 while start <= end and exclude_prefix.search(tb[start]):
2030 start += 1
2031 while start <= end and exclude_suffix.search(tb[end]):
2032 end -= 1
2033 return tb[start : end + 1]
2036def attrsetter(attrname):
2037 code = "def set(obj, value): obj.%s = value" % attrname
2038 env = locals().copy()
2039 exec(code, env)
2040 return env["set"]
2043dunders_re = re.compile("^__.+__$")
2046class TypingOnly:
2047 """A mixin class that marks a class as 'typing only', meaning it has
2048 absolutely no methods, attributes, or runtime functionality whatsoever.
2050 """
2052 __slots__ = ()
2054 def __init_subclass__(cls, **kw: Any) -> None:
2055 if TypingOnly in cls.__bases__:
2056 remaining = {
2057 name for name in cls.__dict__ if not dunders_re.match(name)
2058 }
2059 if remaining:
2060 raise AssertionError(
2061 f"Class {cls} directly inherits TypingOnly but has "
2062 f"additional attributes {remaining}."
2063 )
2064 super().__init_subclass__(**kw)
2067class EnsureKWArg:
2068 r"""Apply translation of functions to accept \**kw arguments if they
2069 don't already.
2071 Used to ensure cross-compatibility with third party legacy code, for things
2072 like compiler visit methods that need to accept ``**kw`` arguments,
2073 but may have been copied from old code that didn't accept them.
2075 """
2077 ensure_kwarg: str
2078 """a regular expression that indicates method names for which the method
2079 should accept ``**kw`` arguments.
2081 The class will scan for methods matching the name template and decorate
2082 them if necessary to ensure ``**kw`` parameters are accepted.
2084 """
2086 def __init_subclass__(cls) -> None:
2087 fn_reg = cls.ensure_kwarg
2088 clsdict = cls.__dict__
2089 if fn_reg:
2090 for key in clsdict:
2091 m = re.match(fn_reg, key)
2092 if m:
2093 fn = clsdict[key]
2094 spec = compat.inspect_getfullargspec(fn)
2095 if not spec.varkw:
2096 wrapped = cls._wrap_w_kw(fn)
2097 setattr(cls, key, wrapped)
2098 super().__init_subclass__()
2100 @classmethod
2101 def _wrap_w_kw(cls, fn: Callable[..., Any]) -> Callable[..., Any]:
2102 def wrap(*arg: Any, **kw: Any) -> Any:
2103 return fn(*arg)
2105 return update_wrapper(wrap, fn)
2108def wrap_callable(wrapper, fn):
2109 """Augment functools.update_wrapper() to work with objects with
2110 a ``__call__()`` method.
2112 :param fn:
2113 object with __call__ method
2115 """
2116 if hasattr(fn, "__name__"):
2117 return update_wrapper(wrapper, fn)
2118 else:
2119 _f = wrapper
2120 _f.__name__ = fn.__class__.__name__
2121 if hasattr(fn, "__module__"):
2122 _f.__module__ = fn.__module__
2124 if hasattr(fn.__call__, "__doc__") and fn.__call__.__doc__:
2125 _f.__doc__ = fn.__call__.__doc__
2126 elif fn.__doc__:
2127 _f.__doc__ = fn.__doc__
2129 return _f
2132def quoted_token_parser(value):
2133 """Parse a dotted identifier with accommodation for quoted names.
2135 Includes support for SQL-style double quotes as a literal character.
2137 E.g.::
2139 >>> quoted_token_parser("name")
2140 ["name"]
2141 >>> quoted_token_parser("schema.name")
2142 ["schema", "name"]
2143 >>> quoted_token_parser('"Schema"."Name"')
2144 ['Schema', 'Name']
2145 >>> quoted_token_parser('"Schema"."Name""Foo"')
2146 ['Schema', 'Name""Foo']
2148 """
2150 if '"' not in value:
2151 return value.split(".")
2153 # 0 = outside of quotes
2154 # 1 = inside of quotes
2155 state = 0
2156 result: List[List[str]] = [[]]
2157 idx = 0
2158 lv = len(value)
2159 while idx < lv:
2160 char = value[idx]
2161 if char == '"':
2162 if state == 1 and idx < lv - 1 and value[idx + 1] == '"':
2163 result[-1].append('"')
2164 idx += 1
2165 else:
2166 state ^= 1
2167 elif char == "." and state == 0:
2168 result.append([])
2169 else:
2170 result[-1].append(char)
2171 idx += 1
2173 return ["".join(token) for token in result]
2176def add_parameter_text(params: Any, text: str) -> Callable[[_F], _F]:
2177 params = _collections.to_list(params)
2179 def decorate(fn):
2180 doc = fn.__doc__ is not None and fn.__doc__ or ""
2181 if doc:
2182 doc = inject_param_text(doc, {param: text for param in params})
2183 fn.__doc__ = doc
2184 return fn
2186 return decorate
2189def _dedent_docstring(text: str) -> str:
2190 split_text = text.split("\n", 1)
2191 if len(split_text) == 1:
2192 return text
2193 else:
2194 firstline, remaining = split_text
2195 if not firstline.startswith(" "):
2196 return firstline + "\n" + textwrap.dedent(remaining)
2197 else:
2198 return textwrap.dedent(text)
2201def inject_docstring_text(
2202 given_doctext: Optional[str], injecttext: str, pos: int
2203) -> str:
2204 doctext: str = _dedent_docstring(given_doctext or "")
2205 lines = doctext.split("\n")
2206 if len(lines) == 1:
2207 lines.append("")
2208 injectlines = textwrap.dedent(injecttext).split("\n")
2209 if injectlines[0]:
2210 injectlines.insert(0, "")
2212 blanks = [num for num, line in enumerate(lines) if not line.strip()]
2213 blanks.insert(0, 0)
2215 inject_pos = blanks[min(pos, len(blanks) - 1)]
2217 lines = lines[0:inject_pos] + injectlines + lines[inject_pos:]
2218 return "\n".join(lines)
2221_param_reg = re.compile(r"(\s+):param (.+?):")
2224def inject_param_text(doctext: str, inject_params: Dict[str, str]) -> str:
2225 doclines = collections.deque(doctext.splitlines())
2226 lines = []
2228 # TODO: this is not working for params like ":param case_sensitive=True:"
2230 to_inject = None
2231 while doclines:
2232 line = doclines.popleft()
2234 m = _param_reg.match(line)
2236 if to_inject is None:
2237 if m:
2238 param = m.group(2).lstrip("*")
2239 if param in inject_params:
2240 # default indent to that of :param: plus one
2241 indent = " " * len(m.group(1)) + " "
2243 # but if the next line has text, use that line's
2244 # indentation
2245 if doclines:
2246 m2 = re.match(r"(\s+)\S", doclines[0])
2247 if m2:
2248 indent = " " * len(m2.group(1))
2250 to_inject = indent + inject_params[param]
2251 elif m:
2252 lines.extend(["\n", to_inject, "\n"])
2253 to_inject = None
2254 elif not line.rstrip():
2255 lines.extend([line, to_inject, "\n"])
2256 to_inject = None
2257 elif line.endswith("::"):
2258 # TODO: this still won't cover if the code example itself has
2259 # blank lines in it, need to detect those via indentation.
2260 lines.extend([line, doclines.popleft()])
2261 continue
2262 lines.append(line)
2264 return "\n".join(lines)
2267def repr_tuple_names(names: List[str]) -> Optional[str]:
2268 """Trims a list of strings from the middle and return a string of up to
2269 four elements. Strings greater than 11 characters will be truncated"""
2270 if len(names) == 0:
2271 return None
2272 flag = len(names) <= 4
2273 names = names[0:4] if flag else names[0:3] + names[-1:]
2274 res = ["%s.." % name[:11] if len(name) > 11 else name for name in names]
2275 if flag:
2276 return ", ".join(res)
2277 else:
2278 return "%s, ..., %s" % (", ".join(res[0:3]), res[-1])
2281def has_compiled_ext(raise_=False):
2282 from ._has_cython import HAS_CYEXTENSION
2284 if HAS_CYEXTENSION:
2285 return True
2286 elif raise_:
2287 raise ImportError(
2288 "cython extensions were expected to be installed, "
2289 "but are not present"
2290 )
2291 else:
2292 return False
2295def load_uncompiled_module(module: _M) -> _M:
2296 """Load the non-compied version of a module that is also
2297 compiled with cython.
2298 """
2299 full_name = module.__name__
2300 assert module.__spec__
2301 parent_name = module.__spec__.parent
2302 assert parent_name
2303 parent_module = sys.modules[parent_name]
2304 assert parent_module.__spec__
2305 package_path = parent_module.__spec__.origin
2306 assert package_path and package_path.endswith("__init__.py")
2308 name = full_name.split(".")[-1]
2309 module_path = package_path.replace("__init__.py", f"{name}.py")
2311 py_spec = importlib.util.spec_from_file_location(full_name, module_path)
2312 assert py_spec
2313 py_module = importlib.util.module_from_spec(py_spec)
2314 assert py_spec.loader
2315 py_spec.loader.exec_module(py_module)
2316 return cast(_M, py_module)
2319_pre_release_normalize = {
2320 "a": "a",
2321 "alpha": "a",
2322 "b": "b",
2323 "beta": "b",
2324 "c": "rc",
2325 "pre": "rc",
2326 "preview": "rc",
2327 "rc": "rc",
2328}
2330_version_string_re = re.compile(
2331 r"""
2332 \s*
2333 (?:[a-z][a-z0-9]*[-_])? # ignored prefix, "py3-"
2334 v?
2335 (?P<release>\d+(?:\.\d+)*)
2336 (?: # pre-release
2337 [-_.]?
2338 (?P<pre_l>alpha|beta|preview|pre|rc|a|b|c)
2339 [-_.]?
2340 (?P<pre_n>\d+)?
2341 )?
2342 (?: # post-release
2343 [-_.]?
2344 (?P<post_l>post|rev|r)
2345 [-_.]?
2346 (?P<post_n>\d+)?
2347 )?
2348 (?: # developmental release
2349 [-_.]?
2350 (?P<dev_l>dev)
2351 [-_.]?
2352 (?P<dev_n>\d+)?
2353 )?
2354 """,
2355 re.X | re.I,
2356)
2358_VersionSortKey = Tuple[
2359 Tuple[int, ...],
2360 Tuple[int, str, int],
2361 Tuple[int, int],
2362 Tuple[int, int],
2363]
2366def _version_sort_key(
2367 release: Tuple[int, ...],
2368 pre: Optional[Tuple[str, int]],
2369 post: Optional[int],
2370 dev: Optional[int],
2371) -> _VersionSortKey:
2372 if pre is None and post is None and dev is not None:
2373 # a dev release with no other qualifiers precedes every
2374 # pre-release of the same release number
2375 pre_key = (-1, "", 0)
2376 elif pre is None:
2377 pre_key = (1, "", 0)
2378 else:
2379 pre_key = (0, pre[0], pre[1])
2381 return (
2382 release,
2383 pre_key,
2384 (0, 0) if post is None else (1, post),
2385 (1, 0) if dev is None else (0, dev),
2386 )
2389def _version_comparison(
2390 op: Callable[[Any, Any], bool],
2391) -> Callable[[VersionInfo, Any], Any]:
2392 """Build one of :class:`.VersionInfo`'s comparison methods.
2394 Comparison takes place against the sort key rather than the tuple
2395 itself, so that pre-release and similar qualifiers are taken into
2396 account. A plain tuple is interpreted as the release segment of a
2397 final release; anything else is not comparable.
2399 """
2401 def compare(self: VersionInfo, other: Any) -> Any:
2402 if isinstance(other, VersionInfo):
2403 other_key = other._sort_key
2404 elif isinstance(other, tuple):
2405 other_key = _version_sort_key(other, None, None, None)
2406 else:
2407 return NotImplemented
2408 return op(self._sort_key, other_key)
2410 return compare
2413class VersionInfo(Tuple[int, ...]):
2414 """A version number, as a tuple of integers.
2416 :class:`.VersionInfo` is a ``tuple`` subclass consisting of the
2417 numeric "release" segment of a version only, e.g. ``2.0.0rc1``
2418 is the tuple ``(2, 0, 0)``. Ordering however takes any
2419 pre-release, post-release and developmental qualifiers into account
2420 as described by :pep:`440`, so that ``2.0.0rc1`` compares as less than
2421 ``2.0.0``, including when compared against a plain tuple such as
2422 ``(2, 0, 0)``.
2424 Plain tuples are interpreted as final releases when compared against
2425 a :class:`.VersionInfo`.
2427 .. versionadded:: 2.1
2429 """
2431 string: Optional[str]
2432 """the string from which this version was parsed, if any."""
2434 pre: Optional[Tuple[str, int]]
2435 """normalized pre-release qualifier, e.g. ``("rc", 1)``."""
2437 post: Optional[int]
2438 """post-release number, if any."""
2440 dev: Optional[int]
2441 """developmental release number, if any."""
2443 _sort_key: _VersionSortKey
2445 def __new__(
2446 cls,
2447 release: Sequence[int] = (),
2448 *,
2449 string: Optional[str] = None,
2450 pre: Optional[Tuple[str, int]] = None,
2451 post: Optional[int] = None,
2452 dev: Optional[int] = None,
2453 ) -> VersionInfo:
2454 # __new__ is needed as the release segment has to be passed to
2455 # tuple.__new__(); the remaining state is set up in __init__
2456 return tuple.__new__(cls, release)
2458 def __init__(
2459 self,
2460 release: Sequence[int] = (),
2461 *,
2462 string: Optional[str] = None,
2463 pre: Optional[Tuple[str, int]] = None,
2464 post: Optional[int] = None,
2465 dev: Optional[int] = None,
2466 ):
2467 self.string = string
2468 self.pre = pre
2469 self.post = post
2470 self.dev = dev
2471 self._sort_key = _version_sort_key(tuple(self), pre, post, dev)
2473 def __repr__(self) -> str:
2474 if self.string is not None:
2475 return f"VersionInfo({tuple(self)!r}, string={self.string!r})"
2476 else:
2477 return f"VersionInfo({tuple(self)!r})"
2479 def __str__(self) -> str:
2480 if self.string is not None:
2481 return self.string
2482 else:
2483 return ".".join(str(num) for num in self)
2485 # every comparison has to be stated explicitly; ``tuple`` implements
2486 # all six of them, so ``functools.total_ordering`` fills in nothing
2487 # here and the ones left out would silently compare as plain tuples
2488 __eq__ = _version_comparison(operator.eq)
2489 __ne__ = _version_comparison(operator.ne)
2490 __lt__ = _version_comparison(operator.lt)
2491 __le__ = _version_comparison(operator.le)
2492 __gt__ = _version_comparison(operator.gt)
2493 __ge__ = _version_comparison(operator.ge)
2495 def __hash__(self) -> int:
2496 return hash(self._sort_key)
2499def parse_version_string(version: Optional[str]) -> VersionInfo:
2500 """Parse a DBAPI version string into a :class:`.VersionInfo`.
2502 Leading characters that are not part of the version itself are
2503 ignored, as are trailing characters following the version, so that
2504 strings such as ``"py3-4.0.19-beta4"`` and
2505 ``"2.9.10 (dt dec pq3 ext lo64)"`` parse correctly.
2507 An empty :class:`.VersionInfo` is returned if no version number can be
2508 located at all.
2510 Parsing is deliberately more tolerant than that of :pep:`440`, which
2511 the version strings published by DBAPIs frequently do not conform to;
2512 a strict implementation such as that of the ``packaging`` library
2513 rejects each of the above outright.
2515 .. versionadded:: 2.1
2517 """
2519 if not version:
2520 return VersionInfo((), string=version)
2522 m = _version_string_re.match(version)
2523 if m is None:
2524 return VersionInfo((), string=version)
2526 release = tuple(int(x) for x in m.group("release").split("."))
2528 pre_l = m.group("pre_l")
2529 pre: Optional[Tuple[str, int]]
2530 if pre_l is not None:
2531 pre = (
2532 _pre_release_normalize[pre_l.lower()],
2533 int(m.group("pre_n") or 0),
2534 )
2535 else:
2536 pre = None
2538 return VersionInfo(
2539 release,
2540 string=version,
2541 pre=pre,
2542 post=(
2543 int(m.group("post_n") or 0)
2544 if m.group("post_l") is not None
2545 else None
2546 ),
2547 dev=(
2548 int(m.group("dev_n") or 0)
2549 if m.group("dev_l") is not None
2550 else None
2551 ),
2552 )
2555def parse_version_from_metadata(distribution: str) -> VersionInfo:
2556 """Return the version of an installed distribution as a
2557 :class:`.VersionInfo`.
2559 This is intended for use by dialects whose DBAPI module does not
2560 itself publish a version number, such as ``asyncmy``. As the
2561 distribution name is not necessarily the same as the module name, and
2562 the installed distribution is not necessarily the module that was
2563 imported, this should not be used when the DBAPI module provides a
2564 version of its own.
2566 An empty :class:`.VersionInfo` is returned if the distribution is not
2567 installed.
2569 .. versionadded:: 2.1
2571 """
2573 try:
2574 version = importlib.metadata.version(distribution)
2575 except importlib.metadata.PackageNotFoundError:
2576 return VersionInfo()
2577 else:
2578 return parse_version_string(version)
2581class _Missing(enum.Enum):
2582 Missing = enum.auto()
2585Missing = _Missing.Missing
2586MissingOr = Union[_T, Literal[_Missing.Missing]]