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