如何为async函数编写非阻塞的entry/exit装饰器以记录完成状态

我需要给async函数编写一个等效的装饰器/包装器，要求满足以下条件：

• 仅在异步函数完成（return或抛出异常）时打印"returned"或"raised"日志
• 不能通过await将异步函数转为同步，装饰器调用时不阻塞
• 对函数体和调用者保持完全透明，行为和普通函数的装饰器一致

现有适用于非异步函数的trace_entry_exit装饰器，用在async函数上会立刻打印"returned"——因为它直接返回了可等待对象，没有等待函数实际执行完成：

import functools

def trace_entry_exit(func):
    """非异步函数的日志追踪装饰器"""
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        print("entry")
        try:
            result = func(*args, **kwargs)
            print("returned")  # 如果func是async函数，这行会立刻执行
        except Exception as ex:
            print("raised")
            raise
        return result  # 直接返回可等待对象，不会等待函数完成
    return wrapper

解决方案：兼容异步/同步的通用装饰器 #

我们可以编写一个能自动识别函数类型的装饰器，对同步函数沿用原有逻辑，对异步函数则使用异步wrapper，内部await原函数以等待执行完成后再打印日志：

import functools
import asyncio

def trace_entry_exit(func):
    """兼容同步和异步函数的日志追踪装饰器"""
    # 同步函数的处理逻辑
    @functools.wraps(func)
    def sync_wrapper(*args, **kwargs):
        print("entry")
        try:
            result = func(*args, **kwargs)
            print("returned")
            return result
        except Exception as ex:
            print("raised")
            raise

    # 异步函数的处理逻辑
    @functools.wraps(func)
    async def async_wrapper(*args, **kwargs):
        print("entry")
        try:
            result = await func(*args, **kwargs)
            print("returned")
            return result
        except Exception as ex:
            print("raised")
            raise

    # 根据函数类型返回对应的wrapper
    return async_wrapper if asyncio.iscoroutinefunction(func) else sync_wrapper

测试验证 #

用以下代码测试异步函数的使用场景：

@trace_entry_exit
async def async_success():
    await asyncio.sleep(1)
    return "异步执行完成"

@trace_entry_exit
async def async_fail():
    await asyncio.sleep(1)
    raise ValueError("异步函数抛出异常")

async def main():
    print("===== 测试正常异步函数 =====")
    print("创建任务，不阻塞主线程")
    task = asyncio.create_task(async_success())
    result = await task
    print(f"返回结果：{result}")

    print("\n===== 测试异常异步函数 =====")
    print("创建任务，不阻塞主线程")
    try:
        task = asyncio.create_task(async_fail())
        await task
    except ValueError as e:
        print(f"捕获异常：{e}")

asyncio.run(main())

运行结果 #

===== 测试正常异步函数 =====
创建任务，不阻塞主线程
entry
returned
返回结果：异步执行完成

===== 测试异常异步函数 =====
创建任务，不阻塞主线程
entry
raised
捕获异常：异步函数抛出异常

可以看到：

• "entry"在函数调用时立刻打印
• "returned"/"raised"仅在异步函数执行完成（sleep 1秒后）才打印
• 创建任务时不会阻塞主线程，完全符合需求

内容的提问来源于stack exchange，提问作者some bits flipped