本文介绍一种巧妙利用协程 send() 机制的 Python 装饰器实现，仅用一套核心逻辑即可统一处理 def 和 async def 函数，彻底消除传统双分支写法中的代码重复，并保持类型安全与运行时正确性。
本文介绍一种巧妙利用协程 `send()` 机制的 python 装饰器实现，仅用一套核心逻辑即可统一处理 `def` 和 `async def` 函数，彻底消除传统双分支写法中的代码重复，并保持类型安全与运行时正确性。
在 Python 中为同步与异步函数编写通用装饰器时，常见方案需分别定义 def decorated(...) 和 async def decorated(...)，导致“前置逻辑”“后置逻辑”等核心行为被复制两次——不仅违反 DRY 原则，更易引入维护偏差与类型注解不一致问题。
本文提供一个真正零逻辑重复的解决方案：统一使用 async def 定义装饰器内层函数，并通过 coroutine.send(None) 同步驱动协程完成，绕过事件循环依赖。其核心思想源于 CPython 社区对协程底层行为的洞察：任何 async def 函数返回的协程对象是可迭代的，调用 .send(None) 即可启动并执行至首个 await（若无 await 则直接完成并抛出 StopIteration）。
以下是完整、生产就绪的实现：
import asyncio
import typing as t
import functools
import time
R = t.TypeVar("R")
P = t.ParamSpec("P")
def is_coroutine(
func: t.Callable[P, t.Any]
) -> t.TypeGuard[t.Callable[P, t.Coroutine[t.Any, t.Any, t.Any]]]:
"""精确判断 func 是否为协程函数（支持 functools.partial 和 async __call__）"""
while isinstance(func, functools.partial):
func = func.func
if asyncio.iscoroutinefunction(func):
return True
if hasattr(func, "__call__"):
# 检查是否为支持 async __call__ 的类实例
call_method = getattr(func, "__call__", None)
if call_method is not None and asyncio.iscoroutinefunction(call_method):
return True
return False
def decorator(func: t.Callable[P, R]) -> t.Callable[P, R]:
is_sync = not is_coroutine(func)
@functools.wraps(func)
async def decorated(*args: P.args, **kwargs: P.kwargs) -> R:
print("Before")
result = func(*args, **kwargs)
if not is_sync:
result = await t.cast(t.Awaitable[R], result)
print(f"After with {result=}")
return result
if is_sync:
# 同步路径：用 send(None) 驱动协程，捕获 StopIteration 获取返回值
@functools.wraps(func)
def decorated_sync(*args: P.args, **kwargs: P.kwargs) -> R:
coro = decorated(*args, **kwargs)
try:
coro.send(None) # 启动协程
except StopIteration as e:
return t.cast(R, e.value) # 成功返回值
raise RuntimeError("协程未正常结束：请勿在同步路径中使用 asyncio.* 或 await") # mypy 兼容占位
return decorated_sync
return decorated # 异步路径直接返回协程函数✅ 关键优势：
- ✅ 零业务逻辑重复：Before / result = ... / After 等所有装饰逻辑仅书写一次，位于 async def decorated 内；
- ✅ 类型安全：借助 TypeGuard 和 ParamSpec + TypeVar，完美保留原始函数签名与返回类型；
- ✅ 无缝兼容：支持普通函数、async def、带 functools.partial 的函数，甚至含 async __call__ 的可调用对象。
⚠️ 重要限制与注意事项：
- 同步调用路径（即 decorated_sync）不可访问事件循环：asyncio.sleep()、await aiohttp.get() 等均会崩溃。所有异步操作必须严格限定在 if not is_sync: 分支内；
- 推荐仅在装饰器的“横切逻辑”（如日志、计时、参数校验、结果包装）中使用此模式；若需复杂异步 I/O，应明确区分 sync/async 装饰器；
- 生产环境建议增加 @overload 声明或 typing.runtime_checkable 协议增强 IDE 支持。
使用示例：
@decorator
def sync_call() -> str:
time.sleep(0.1)
return "sync"
@decorator
async def async_call() -> str:
await asyncio.sleep(0.1)
return "async"
# 在异步上下文中混合调用
async def main():
print(sync_call()) # → "sync"
print(await async_call()) # → "async"
asyncio.run(main())输出清晰印证双路径正确性：
Before After with result='sync' sync Before After with result='async' async
该方案以少量底层协程知识为代价，换取了装饰器逻辑的极致简洁与可维护性——是 Python 高级装饰器设计中兼具巧思与实用性的典范实践。
