如何编写无代码重复的Python同步/异步通用装饰器？

你提到的这种兼容同步def和异步async def的装饰器，确实容易出现前置/后置逻辑代码重复的问题——核心矛盾就是同步和异步分支里的业务逻辑完全一致，只是调用目标函数的方式不同（直接调用 vs await）。

我们可以通过抽离公共逻辑或者统一异步适配两种方式来消除代码重复，下面是具体实现：

方案一：抽离公共前置/后置逻辑函数 #

把重复的前置、后置逻辑单独封装成函数，同步和异步分支只需要调用这些公共函数，再处理各自的函数调用方式即可：

import asyncio
import typing as t
import functools

R = t.TypeVar("R")
P = t.ParamSpec("P")

# 抽离公共前置逻辑
def _before(*args, **kwargs):
    # 示例：打印调用参数、启动计时器等
    print(f"准备执行函数，参数：{args} | {kwargs}")

# 抽离公共后置逻辑
def _after(result: R) -> R:
    # 示例：处理结果、记录执行时长等
    print(f"函数执行完成，结果：{result}")
    return result

def decorator(func: t.Callable[P, R]) -> t.Callable[P, R]:
    if asyncio.iscoroutinefunction(func):
        @functools.wraps(func)
        async def decorated(*args: P.args, **kwargs: P.kwargs) -> R:
            _before(*args, **kwargs)
            result = await func(*args, **kwargs)
            return _after(result)
    else:
        @functools.wraps(func)
        def decorated(*args: P.args, **kwargs: P.kwargs) -> R:
            _before(*args, **kwargs)
            result = func(*args, **kwargs)
            return _after(result)
        
    return decorated

这种方式逻辑清晰，修改公共逻辑时只需要改动_before和_after，不需要同步修改两个分支的代码。

方案二：统一异步适配逻辑 #

先把同步函数转换为异步函数，只编写一套异步装饰逻辑，最后再根据原函数类型返回对应的同步/异步装饰结果：

import asyncio
import typing as t
import functools

R = t.TypeVar("R")
P = t.ParamSpec("P")

# 同步函数转异步适配器
def _sync_to_async(func: t.Callable[P, R]) -> t.Callable[P, t.Awaitable[R]]:
    @functools.wraps(func)
    async def wrapper(*args: P.args, **kwargs: P.kwargs) -> R:
        return func(*args, **kwargs)
    return wrapper

def decorator(func: t.Callable[P, R | t.Awaitable[R]]) -> t.Callable[P, R | t.Awaitable[R]]:
    # 统一将目标函数转为异步类型
    target_func = _sync_to_async(func) if not asyncio.iscoroutinefunction(func) else func

    # 只编写一次异步装饰逻辑
    @functools.wraps(func)
    async def async_decorated(*args: P.args, **kwargs: P.kwargs) -> R:
        # 公共前置逻辑
        print("执行前置操作...")
        result = await target_func(*args, **kwargs)
        # 公共后置逻辑
        print("执行后置操作...")
        return result
    
    # 根据原函数类型返回对应装饰结果
    if asyncio.iscoroutinefunction(func):
        return async_decorated
    else:
        @functools.wraps(func)
        def sync_decorated(*args: P.args, **kwargs: P.kwargs) -> R:
            # 同步环境下运行异步装饰逻辑
            return asyncio.run(async_decorated(*args, **kwargs))
        return sync_decorated

这种方案适合装饰逻辑复杂的场景，只需要维护一套核心逻辑，避免了分支重复。注意：如果在已有事件循环的异步环境（如FastAPI、Django ASGI）中使用，建议用asyncio.get_running_loop().run_in_executor替代asyncio.run，避免创建新事件循环的冲突。

内容的提问来源于stack exchange，提问作者Kound