1from asyncutils._internal import patch as P
2from asyncutils._internal.helpers import fullname
3from asyncutils._internal.submodules import constants_all as __all__
4import sys as S
5RECIPROCAL_E, EXECUTORS_FROZENSET = 0.36787944117144233, frozenset(POSSIBLE_EXECUTORS := ('thread', 'process', 'interpreter', 'loky', 'loky_no_reuse', 'dask', 'ipython', 'elib_flux_cluster', 'elib_flux_job', 'elib_slurm_cluster', 'elib_slurm_job', 'elib_single_node', 'pebble_thread', 'pebble_process', 'deadpool'))
6if (f := getattr(S, '_getframemodulename', None)) is None: # pragma: no cover
7 if (u := getattr(S, '_getframe', None)) is None: raise SystemError('asyncutils: expected _getframe to be defined in sys')
8 def f(depth=0, _=u): return _(depth).f_globals.get('__name__')
[docs]
9class SentinelBase:
10 _can_instantiate = False; __slots__ = '__mod', '__name'
11 def __new__(cls, name=None, _=__import__('keyword').iskeyword, g=f):
12 cls._assert_can_instantiate()
13 if name is None: return super().__new__(cls)
14 if _(name) or not all(p.isidentifier() and not _(p) for p in name.split('.', 1)): raise ValueError('asyncutils.constants.SentinelBase: invalid name')
15 if (m := g(1)) is not None: name = f'{m}.{name}'
16 if (o := (c := cls._cache).get(name)) is None:
17 (o := super().__new__(cls)).__name, o.__mod = name, m
18 with cls._lock: c[name] = o
19 return o
20 @property
21 def name(self): return self.__name
22 @property
23 def module(self): return self.__mod # ty: ignore[unresolved-attribute]
24 @classmethod
25 def _assert_can_instantiate(cls):
26 if not cls._can_instantiate: raise TypeError(f'cannot instantiate {fullname(cls)!r}')
27 def __repr__(self): return f'<{fullname(self)} {self.__name!r} at {id(self):#x}>'
28 def __str__(self): return getattr(self, '_SentinelBase__name', '<unbound>')+(' <private>'*self.is_private)
[docs]
29 def __set_name__(self, owner, name, /, _='NOTE: The following is not allowed:\nclass {0}:\n{1} = {2}({3!r})\n...\ninstead, use:\nclass {0}:\n{1} = {2}()\n'.format):
30 N = f'{fullname(owner)}.{name}'.replace('<locals>.', '').replace('<lambda>.', '')
31 with self._lock:
32 if (n := getattr(self, '_SentinelBase__name', None)) is None:
33 if self is not self._cache.setdefault(N, self): raise NameError(f'{fullname(self)} name collision', name=N)
34 self.__name = N
35 elif n != N or self._cache.get(N) is not self: raise NameError(f'cannot bind named {fullname(self)} to class {fullname(owner)!r}', name=N)
36 if (b := self.bound_to) is None: raise NameError(f'cannot bind named unbound {fullname(self)} to class {fullname(owner)!r}', name=N)
37 __import__('sys').stderr.write(_(b.rpartition('.')[-1], self.back, fullname(self), N))
[docs]
38 def __reduce__(self):
39 if (n := getattr(self, '_SentinelBase__name', None)) is None: raise TypeError(f'cannot pickle unbound instance of {fullname(self)}')
40 return type(self), (n,)
[docs]
41 def __init_subclass__(cls, *, lock_impl=__import__('_thread').allocate_lock):
42 if cls.__dict__.get('__slots__', True): raise TypeError('asyncutils.constants.SentinelBase: sentinel classes should have empty __slots__')
43 cls._cache, cls._lock, cls._can_instantiate = {}, lock_impl(), True
44 @property
45 def is_private(self): return (self.back or '').startswith('_')
46 @property
47 def bound_to(self): return getattr(self, '_SentinelBase__name', '').rpartition('.')[0] or None
48 @property
49 def back(self): return getattr(self, '_SentinelBase__name', '').rpartition('.')[2] or None
50 P.patch_classmethod_signatures((__new__, 'name=None'), (__init_subclass__, 'lock_impl={}')); P.patch_method_signatures((__set_name__, 'owner, name, /'))
51class _Sentinel(SentinelBase):
52 __slots__ = ()
53 def __init_subclass__(cls, /, **_): raise TypeError('cannot subclass the type of asyncutils-internal sentinels')
54 def __reduce__(self): return self.back
55_NO_DEFAULT, RAISE, NO_COALESCE = map(_Sentinel, ('_NO_DEFAULT', 'RAISE', 'NO_COALESCE'))
56_Sentinel._can_instantiate = False
57del _Sentinel, P