异步函数的入口/出口装饰器

Question

我如何为async函数编写一个等效的装饰器/包装器，它将在异步完成(returns或raises(时进行日志记录，而不简单地通过awaits使其同步。

import functools
def trace_entry_exit(func):
"""non asyncio decorator"""
@functools.wraps(func)
def wrapper(*args, **kwargs):
print("entry")
try:
result = func(*args, **kwargs)
print("returned")  ### if func is async, this will print ~immediately
except Exception as ex:
print("raised")
raise
return result ### without awaiting func(), this will simply return the awaitable object ~immediately
return wrapper

我希望returned或raised只在函数返回(或引发(时打印，但在decorator中的调用上打印而不是块。

这同样适用于任何包装func的东西，只是decorator用例是唯一的，因为它对func主体和调用func()的人都是透明的，就像在非异步情况下一样。

Answer 1

只需制作包装async即可。然后你可以await这个函数，它将像包装好的函数一样工作，但在它完成后使用print：

import functools
def trace_entry_exit(func):
"""asyncio decorator"""
@functools.wraps(func)
async def wrapper(*args, **kwargs):
print("entry")
try:
result = await func(*args, **kwargs)
print("returned")
except Exception as ex:
print("raised")
raise
return result ### without awaiting func(), this will simply return the awaitable object ~immediately
return wrapper

如果你想让它为协同程序或同步函数工作：

import functools
import asyncio
def trace_entry_exit(func):
"""asyncio or not decorator"""
if asyncio.iscoroutinefunction(func):
@functools.wraps(func)
async def wrapper(*args, **kwargs):
print("entry")
try:
result = await func(*args, **kwargs)
print("returned")
except Exception as ex:
print("raised")
raise
return result
return wrapper
else:
@functools.wraps(func)
def wrapper(*args, **kwargs):
print("entry")
try:
result = func(*args, **kwargs)
print("returned")
except Exception as ex:
print("raised")
raise
return result
return wrapper