35 lines
1.1 KiB
Python
35 lines
1.1 KiB
Python
# SPDX-License-Identifier: AGPL-3.0-only
|
|
|
|
import random
|
|
import contextlib
|
|
import contextvars
|
|
|
|
class ContextVar:
|
|
"""
|
|
contextvars.ContextVar, but you can call the var to get its value, rather than calling var.get()
|
|
"""
|
|
__slots__ = frozenset({'_var'})
|
|
def __init__(self, *a, **kw): self._var = contextvars.ContextVar(*a, **kw)
|
|
def __call__(self, *args): return self._var.get(*args) # convenience
|
|
def get(self, *args): return self._var.get(*args)
|
|
def set(self, *args): return self._var.set(*args)
|
|
def reset(self, *args): return self._var.reset(*args)
|
|
|
|
class asyncnullcontext(contextlib.AbstractAsyncContextManager):
|
|
async def __aenter__(self): return None
|
|
async def __aexit__(self, *excinfo): return None
|
|
|
|
def build_dprint(debug: bool):
|
|
if debug:
|
|
class dprint(asyncnullcontext):
|
|
# 8 byte pseudo-random request tag
|
|
def __init__(self): self.req_id = ''.join(random.choices('0123456789abcdef', k=8 * 2))
|
|
def print(self, *args, **kwargs): print(f'[{self.req_id}]', *args, **kwargs)
|
|
async def __aenter__(self): return self.print
|
|
else:
|
|
class dprint(asyncnullcontext):
|
|
async def __aenter__(self):
|
|
return lambda *args, **kwargs: None
|
|
|
|
return dprint
|