1import asyncutils._internal.patch as P, sys as S
[docs]
2def filter_out(*a, s=None): yield from filter(lambda x, s=s: s is not x, a)
[docs]
3def get_loop_and_set(_=(lambda l: l.stop() or l.close()).__get__, f=__import__('atexit').register):
4 import asyncio.events as E
5 if (l := E._get_running_loop()) is None: f(_(l := E.new_event_loop())); E.set_event_loop(l)
6 S.audit('asyncutils/get_loop_and_set', l); return l
[docs]
7def check_methods(obj, /, *meth):
8 M = obj.__class__.__mro__
9 for m in meth:
10 for b in M:
11 if (_ := b.__dict__.get(m, obj)) is None: return False
12 if _ is not obj: break
13 else: return False
14 return True
[docs]
15def copy_and_clear(l): r = l.copy(); l.clear(); return r
[docs]
16def ismodule(o, /, _=frozenset(('asyncutils._internal.initialize.Module', 'builtins.module'))): return fullname(type(o)) in _
[docs]
17def subscriptable(cls, /, _=classmethod(type(list[int]))):
18 if hasattr(cls, '__class_getitem__'): raise TypeError('class is already subscriptable')
19 cls.__class_getitem__ = _; return cls
[docs]
20def check(a, b, /): return a is b or (False if (e := b.__eq__(a)) is NotImplemented else e) # noqa: PLC2801
[docs]
21def coerce_callable(o, /): return o if callable(o) else type(o)
[docs]
22def create_executor(f, /, save=True): # pragma: no cover
23 S.audit('asyncutils/create_executor', f'{(F := coerce_callable(f)).__module__.removeprefix('asyncutils.')}.{F.__qualname__}'); from asyncutils import Executor; e = Executor()
24 if save: f.executor = e
25 return e
[docs]
26def fullname(f, /, remove_prefix=False, _=('__module__', '__qualname__')): f = coerce_callable(f); n = '.'.join(filter_out(*(getattr(f, a, None) for a in _))); return n.removeprefix('asyncutils.') if remove_prefix else n
[docs]
27async def simple_wrap(aw, /): return await aw
[docs]
28class LoopMixinBase:
29 __slots__ = '_loop',
30 @property
31 def loop(self):
32 if (l := getattr(self, '_loop', None)) is None: self._loop = l = get_loop_and_set()
33 elif l is not __import__('asyncio.events', fromlist=('',))._get_running_loop(): raise RuntimeError('asyncutils: could not bind asyncio event loop')
34 return l
[docs]
35 def make(self, a, /): return self.loop.create_task(simple_wrap(a))
[docs]
36 def make_fut(self): return self.loop.create_future()
[docs]
37 def make_multiple(self, a, /): yield from map(self.make, a)
[docs]
38class Bag(dict): # noqa: FURB189
39 __slots__, __setattr__, __delattr__ = (), dict.__setitem__, dict.__delitem__
[docs]
40 def __getattr__(self, k, /):
41 try: return self[k]
42 except KeyError: raise AttributeError(name=k, obj=self) from None
43P.patch_function_signatures((ismodule, 'o, /'), (subscriptable, 'cls, /'), (fullname, 'f, /, remove_prefix=False'), (get_loop_and_set, ''))
44del P