1# ty: ignore[unresolved-reference]
2from asyncutils._internal import patch as P
3from asyncutils._internal.compat import p
4from asyncutils._internal.submodules import context_all as __all__
5from asyncutils._internal.unparsed import C
6_, k, all_contextual_consts = __import__('_contextvars').ContextVar('asyncutils_contextvar'), None, frozenset(C)
[docs]
7class Context: # noqa: PLW1641
8 __slots__ = tuple(C); exec(f'def __new__(cls,/,*,{",".join(f"{k}={v!r}" for k, v in C.items())}):\n\t(_:=object.__new__(cls)).{"\n\t_.".join(f"{k}={k}" for k in __slots__)}\n\treturn _') # noqa: S102
9 def __init_subclass__(cls, /, **_): raise TypeError('cannot subclass asyncutils.context.Context')
10 def __getattribute__(self, n, /, _=frozenset(('ascurctx', 'replace_from_dct', 'replace', 'update', 'asdict', 'copy', 'pprint', 'from_dct')), u='__'): return super().__getattribute__(n if n in _ or (n.startswith(u) and n.endswith(u)) else n.upper())
[docs]
11 def __getitem__(self, n, /): return super().__getattribute__(n.upper())
12 def __setattr__(self, n, v, /):
13 if (n := n.upper()) not in all_contextual_consts: raise AttributeError('asyncutils.context.Context: attribute not found', name=n, obj=self)
14 if isinstance(v, list):
15 if v and isinstance(v[0], list): v = map(tuple, v) # ty: ignore[invalid-argument-type]
16 v = tuple(v)
17 super().__setattr__(n, v)
18 def __delattr__(self, n, /): raise AttributeError('asyncutils.context.Context: cannot delete attribute', name=n, obj=self)
[docs]
19 def replace_from_dct(self, d, /, _=all_contextual_consts):
20 D = self.asdict()
21 for n, v in d.items():
22 if (n := n.upper()) in _: D[n] = v
23 return type(self)(**D)
[docs]
24 def update(self, d=None, _=all_contextual_consts, /, **k):
25 for m in (d, k):
26 if not m: continue
27 for n, v in m.items():
28 if (n := n.upper()) in _: setattr(self, n, v)
[docs]
29 def ascurctx(self, **k): return nonreusablelocalcontext(self, **k)
[docs]
30 def __eq__(self, o, /):
31 if type(o) is not __class__: return False
32 f, g = map(object.__getattribute__.__get__, (self, o)); return all(f(k) == g(k) for k in self.__slots__)
[docs]
33 @classmethod
34 def from_dct(cls, d, /): return cls(**{k.upper(): v for k, v in d.items()})
[docs]
35 def asdict(self): return {k: getattr(self, k) for k in self.__slots__}
[docs]
36 def copy(self): return type(self)(**self.asdict())
[docs]
37 def replace(self, /, **k): return self.replace_from_dct(k)
[docs]
38 def pprint(self, file=__import__('sys').stdout, *, flush=True, pp=None, incl_newline=True): file.write('Context.from_dct(\n'); (pp or p)._format(self.asdict(), file, 0, 0, {}, 0); print('\n)', end='\n'*incl_newline, file=file, flush=flush)
39 def __str__(self, _=__import__('_io').StringIO): self.pprint(s := _(), incl_newline=False); return s.getvalue()
40 def __repr__(self): return f'Context({", ".join(f"{k}={getattr(self, k)!r}" for k in self.__slots__)})'
41 def __reduce__(self): return __class__.from_dct, (self.asdict(),)
42 __copy__, __replace__, __setitem__ = copy, replace, __setattr__; P.patch_method_signatures((__str__, ''), (update, 'd=None, /, **k'), (pprint, 'file={0}, *, pp={0}, incl_newline=True'), (replace_from_dct, 'd, /'), (__getattribute__, 'name, /'))
[docs]
43def getcontext(_=_, d=Context()): # noqa: B008
44 try: return _.get()
45 except LookupError: _.set(d); return d
[docs]
46def setcontext(c, /, _=_):
47 if not isinstance(c, Context): raise TypeError('asyncutils.context.setcontext: ctx must be an instance of asyncutils.context.Context')
48 _.set(c)
[docs]
49class localcontext:
50 __slots__ = 'new_ctx', 'saved_ctx'
51 def __init__(self, /, ctx=None, **k):
52 if ctx is None: ctx = getcontext()
53 if type(ctx) is not Context: raise TypeError('asyncutils.context.localcontext: ctx must be an instance of asyncutils.context.Context')
54 self.new_ctx = ctx.replace_from_dct(k)
[docs]
55 def __enter__(self): self.saved_ctx = getcontext(); setcontext(c := self.new_ctx); return c
[docs]
56 def __exit__(self, /, *_):
57 setcontext(self.saved_ctx); del self.saved_ctx
58 if isinstance(self, nonreusablelocalcontext): del self.new_ctx
[docs]
59 async def __aenter__(self): return self.__enter__()
[docs]
60 async def __aexit__(self, /, *_): return self.__exit__(*_)
61 P.patch_method_signatures((__exit__, s := P.xsig), (__aexit__, s)); del s
[docs]
62class nonreusablelocalcontext(localcontext): __slots__ = ()
63def __getattr__(n, /, _=getcontext): return getattr(_(), n)
64P.patch_function_signatures((getcontext, ''), (setcontext, 'ctx, /'), (__getattr__, 'name, /'))
65del _, P, k, C