Код: Выделить всё
def make_wrapped_awaitable_or_aiter(
awaitable_or_aiter, span_wrapper, ContextManager]
) -> AwaitableOrAIterT:
if isinstance(awaitable_or_aiter, AsyncIterator):
async def inner_iterable():
with span_wrapper():
async for v in awaitable_or_aiter:
yield v
return inner_iterable()
else:
async def inner_awaitable():
with span_wrapper():
return await awaitable_or_aiter
return inner_awaitable()
def decorator_factory(decorator_factory_arg1):
def decorator(original_func):
def wrapped(*args, **kwargs):
return awaitable_or_aiter(original_func(*args, **kwargs), some_contextmanager)
return wrapped
return decorator
@decorator_factory('foo')
def foobar('anything'):
...
Я хотел бы ввести его на Python 3.8, чтобы mypy понимал, что тип возвращаемого значения обернутой функции будет таким же, как тип возвращаемого значения original_func, и что он также будет иметь тот же тип возвращаемого значения, который можно вызвать асинхронно. На данный момент у меня это есть:
Код: Выделить всё
AwaitableOrAIterT = TypeVar("AwaitableOrAIterT", Awaitable, AsyncIterator)
def make_wrapped_awaitable_or_aiter(
awaitable_or_aiter: AwaitableOrAIterT, span_wrapper: Callable[[], ContextManager]
) -> AwaitableOrAIterT:
if isinstance(awaitable_or_aiter, AsyncIterator):
async def inner_iterable():
with span_wrapper():
async for v in awaitable_or_aiter:
yield v
return inner_iterable()
else:
async def inner_awaitable():
with span_wrapper():
return await awaitable_or_aiter
return inner_awaitable()
def decorator_factory(decorator_factory_arg1: str) -> Callable[[Callable[..., AwaitableOrAIterT]], Callable[..., AwaitableOrAIterT]]:
def decorator(original_func: Callable[..., AwaitableOrAIterT]):
def wrapped(*args, **kwargs) -> AwaitableOrAIterT:
make_wrapped_awaitable_or_aiter(original_func(*args, **kwargs), some_contextmanager)
return wrapped
return decorator
@decorator_factory('foo')
async def foobar(some_arg: str) -> bool:
return await some_query_typed_to_bool()
По сути, я считаю, что мне нужен способ параметризовать AwaitableOrAIterT как универсальный (используя возвращаемый тип типа Awaitable или AsyncIterator), чтобы я мог комментировать
Код: Выделить всё
def make_wrapped_awaitable_or_aiter(
awaitable_or_aiter: AwaitableOrAIterT[T], span_wrapper: Callable[[], ContextManager]
) -> AwaitableOrAIterT[T]:
if isinstance(awaitable_or_aiter, AsyncIterator):
async def inner_iterable() -> T:
with span_wrapper():
async for v in awaitable_or_aiter:
yield v
return inner_iterable()
..etc..
Подробнее здесь: https://stackoverflow.com/questions/735 ... python-3-8