Декораторы для корутин в asyncio¶
Декораторы для асинхронных функций пишутся как и для обычных только возвращать нужно корутину, а не функцию.
Для примера обычная функция:
def plusplus(func): def wrapped(): return func() + 1 return wrapped @plusplus def one(): return 1 print(one()) # return 2 @plusplus @plusplus @plusplus @plusplus def one(): return 1 print(one()) # now return 5
В декораторе наша обертка над функцией ( wrapped ) стала корутиной:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37
import timeit import asyncio def plusplus(func): async def wrapped(): await asyncio.sleep(1) return await func() + 1 return wrapped @plusplus async def one(): await asyncio.sleep(2) return 1 loop = asyncio.get_event_loop() start = timeit.default_timer() print(loop.run_until_complete(one())) # return 2 stop = timeit.default_timer() print(stop - start) # minimum 3 seconds @plusplus @plusplus @plusplus @plusplus async def one(): await asyncio.sleep(2) return 1 start = timeit.default_timer() print(loop.run_until_complete(one())) # return 5 stop = timeit.default_timer() print(stop - start) # minimum 6 seconds
Более практичный пример это функция json_response для вьюх в aiohttp. Идея взята из презентации (http://igordavydenko.com/talks/lvivpy-4/#slide-31).
import ujson import asyncio from aiohttp import web def json_response(data, **kwargs): kwargs.setdefault('content_type', 'application/json') return web.Response(text=ujson.dumps(data), **kwargs) async def index(request): return json_response("Hello": "World">)
Все хорошо но ретурнов во вьюхе может быть много и тогда оборачивать каждый в json_response довольно неудобно. Чтобы решить эту проблему создадим декоратор json_view .
def json_view(func): async def wrapped(request): return json_response(await func(request)) return wrapped
@json_view async def index(request): if somethink: return "Somethink": "happens"> else: return "else": "happens"> return "Hello": "World">
Класс aiohttp.web.Response позволяет задавать различные параметры типа заголовков и статуса ответа. Перепишем наш декоратор таким образом чтобы он умел принимать эти параметры:
def json_view_arg(**kwargs): def wrap(func): async def wrapped(request): return json_response(await func(request), **kwargs) return wrapped return wrap
Теперь можно задать, например, кастомный заголовок ответа Server :
@json_view_arg(headers="Server": "Nginx">) async def index(request): return "Hello": "World">
И в заключение то же в виде класса-декоратора:
class JsonView(object): def __init__(self, **kwargs): self.kwargs = kwargs def __call__(self, func): async def wrapped(request): return json_response(await func(request), **self.kwargs) return wrapped
@JsonView(headers="Server": "Nginx">) async def index(request): return "Hello": "World">
© Copyright 2020, Кафедра Интеллектуальных Информационных Технологий ИнФО УрФУ. Created using Sphinx 1.7.6.
Integralist / Python Async Decorator.py
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters. Learn more about bidirectional Unicode characters
| import asyncio |
| import time |
| from functools import wraps |
| def dec ( fn ): |
| if asyncio . iscoroutinefunction ( fn ): |
| @ wraps ( fn ) |
| async def wrapper ( * args , ** kwargs ): |
| print ( fn , args , kwargs ) # () <> |
| await asyncio . sleep ( 5 ) |
| print ( «done with wrapper, going to call fn» ) |
| return await fn () |
| return wrapper |
| else : |
| @ wraps ( fn ) |
| def wrapper ( * args , ** kwargs ): |
| print ( fn , args , kwargs ) # () <> |
| time . sleep ( 5 ) |
| print ( «done with wrapper, going to call fn» ) |
| return fn () |
| return wrapper |
| @ dec |
| async def foo (): |
| return await asyncio . sleep ( 5 , result = «async function done» ) |
| @ dec |
| def bar (): |
| time . sleep ( 5 ) |
| return «sync function done» |
| loop = asyncio . get_event_loop () |
| result = loop . run_until_complete ( foo ()) |
| print ( result ) # async function done |
| print ( bar ()) # sync function done |
| loop . close () |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters. Learn more about bidirectional Unicode characters
| import asyncio |
| from functools import wraps |
| def dec ( fn ): |
| @ wraps ( fn ) |
| async def wrapper ( * args , ** kwargs ): |
| print ( fn , args , kwargs ) # () <> |
| await asyncio . sleep ( 5 ) |
| print ( «done with wrapper, going to call fn» ) |
| return await fn () |
| return wrapper |
| @ dec |
| async def foo (): |
| return await asyncio . sleep ( 5 , result = «i’m done» ) |
| loop = asyncio . get_event_loop () |
| result = loop . run_until_complete ( foo ()) |
| print ( result ) # i’m done |
| loop . close () |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters. Learn more about bidirectional Unicode characters
| import asyncio |
| from functools import wraps |
| async def foo (): |
| return await asyncio . sleep ( 5 , result = «i’m done» ) |
| loop = asyncio . get_event_loop () |
| result = loop . run_until_complete ( foo ()) |
| print ( result ) # i’m done |
| loop . close () |
The proper wrapper args are (*args, **kwargs) and the wrapped function should be called as fn(*args, **kwargs).
Obviously all this print and sleep code wouldn’t be there in the general case either. But I like the idea of one decorator for both async and sync code.
probably your forget to pass args to wrapped fn
Footer
You can’t perform that action at this time.