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