venv/lib/python3.11/site-packages/blinker/_utilities.py

   1 from __future__ import annotations
   2
   3 import asyncio
   4 import inspect
   5 import sys
   6 import typing as t
   7 from functools import partial
   8 from weakref import ref
   9
  10 from blinker._saferef import BoundMethodWeakref
  11
  12 IdentityType = t.Union[t.Tuple[int, int], str, int]
  13
  14
  15 class _symbol:
  16     def __init__(self, name):
  17         """Construct a new named symbol."""
  18         self.__name__ = self.name = name
  19
  20     def __reduce__(self):
  21         return symbol, (self.name,)
  22
  23     def __repr__(self):
  24         return self.name
  25
  26
  27 _symbol.__name__ = "symbol"
  28
  29
  30 class symbol:
  31     """A constant symbol.
  32
  33     >>> symbol('foo') is symbol('foo')
  34     True
  35     >>> symbol('foo')
  36     foo
  37
  38     A slight refinement of the MAGICCOOKIE=object() pattern.  The primary
  39     advantage of symbol() is its repr().  They are also singletons.
  40
  41     Repeated calls of symbol('name') will all return the same instance.
  42
  43     """
  44
  45     symbols = {}  # type: ignore[var-annotated]
  46
  47     def __new__(cls, name):
  48         try:
  49             return cls.symbols[name]
  50         except KeyError:
  51             return cls.symbols.setdefault(name, _symbol(name))
  52
  53
  54 def hashable_identity(obj: object) -> IdentityType:
  55     if hasattr(obj, "__func__"):
  56         return (id(obj.__func__), id(obj.__self__))  # type: ignore[attr-defined]
  57     elif hasattr(obj, "im_func"):
  58         return (id(obj.im_func), id(obj.im_self))  # type: ignore[attr-defined]
  59     elif isinstance(obj, (int, str)):
  60         return obj
  61     else:
  62         return id(obj)
  63
  64
  65 WeakTypes = (ref, BoundMethodWeakref)
  66
  67
  68 class annotatable_weakref(ref):
  69     """A weakref.ref that supports custom instance attributes."""
  70
  71     receiver_id: t.Optional[IdentityType]
  72     sender_id: t.Optional[IdentityType]
  73
  74
  75 def reference(  # type: ignore[no-untyped-def]
  76     object, callback=None, **annotations
  77 ) -> annotatable_weakref:
  78     """Return an annotated weak ref."""
  79     if callable(object):
  80         weak = callable_reference(object, callback)
  81     else:
  82         weak = annotatable_weakref(object, callback)
  83     for key, value in annotations.items():
  84         setattr(weak, key, value)
  85     return weak  # type: ignore[no-any-return]
  86
  87
  88 def callable_reference(object, callback=None):
  89     """Return an annotated weak ref, supporting bound instance methods."""
  90     if hasattr(object, "im_self") and object.im_self is not None:
  91         return BoundMethodWeakref(target=object, on_delete=callback)
  92     elif hasattr(object, "__self__") and object.__self__ is not None:
  93         return BoundMethodWeakref(target=object, on_delete=callback)
  94     return annotatable_weakref(object, callback)
  95
  96
  97 class lazy_property:
  98     """A @property that is only evaluated once."""
  99
 100     def __init__(self, deferred):
 101         self._deferred = deferred
 102         self.__doc__ = deferred.__doc__
 103
 104     def __get__(self, obj, cls):
 105         if obj is None:
 106             return self
 107         value = self._deferred(obj)
 108         setattr(obj, self._deferred.__name__, value)
 109         return value
 110
 111
 112 def is_coroutine_function(func: t.Any) -> bool:
 113     # Python < 3.8 does not correctly determine partially wrapped
 114     # coroutine functions are coroutine functions, hence the need for
 115     # this to exist. Code taken from CPython.
 116     if sys.version_info >= (3, 8):
 117         return asyncio.iscoroutinefunction(func)
 118     else:
 119         # Note that there is something special about the AsyncMock
 120         # such that it isn't determined as a coroutine function
 121         # without an explicit check.
 122         try:
 123             from unittest.mock import AsyncMock  # type: ignore[attr-defined]
 124
 125             if isinstance(func, AsyncMock):
 126                 return True
 127         except ImportError:
 128             # Not testing, no asynctest to import
 129             pass
 130
 131         while inspect.ismethod(func):
 132             func = func.__func__
 133         while isinstance(func, partial):
 134             func = func.func
 135         if not inspect.isfunction(func):
 136             return False
 137
 138         if func.__code__.co_flags & inspect.CO_COROUTINE:
 139             return True
 140
 141         acic = asyncio.coroutines._is_coroutine  # type: ignore[attr-defined]
 142         return getattr(func, "_is_coroutine", None) is acic