本文介绍一种巧妙利用协程 send() 方法特性的装饰器设计，使单个装饰器逻辑同时支持 def 和 async def 函数，彻底避免传统双分支写法中的前后置逻辑重复。
本文介绍一种巧妙利用协程 `send()` 方法特性的装饰器设计，使单个装饰器逻辑同时支持 `def` 和 `async def` 函数，彻底避免传统双分支写法中的前后置逻辑重复。
在 Python 装饰器开发中，一个常见痛点是：为同时兼容同步函数（def）和异步函数（async def），开发者往往被迫编写两套几乎完全相同的逻辑——仅因调用方式不同（result = func(...) vs result = await func(...)），导致“执行前”“执行后”等核心逻辑被重复书写，违背 DRY 原则，也增加维护成本。
上述问题的优雅解法在于统一以异步方式编写装饰器主体，并通过协程的底层机制在同步上下文中安全驱动其执行。关键洞察来自 CPython 的一个实践：async def 函数返回的是一个协程对象（Coroutine），而协程对象是可迭代的——它实现了 __iter__ 和 send() 方法；对刚创建的协程调用 .send(None) 会启动执行，直到遇到第一个 await（若无 await，则直接运行至完成并抛出 StopIteration，其 value 即为返回值）。这意味着：一个纯 async def decorated(...) 可以在不依赖事件循环的前提下，被同步函数安全“驱动”完成。
以下是完整、类型安全、生产可用的实现：
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]]]:
"""精确判断函数是否为协程函数（支持 functools.partial 和 async __call__）"""
# 解包 functools.partial
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 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("Decorator internal error: coroutine did not terminate cleanly")
return decorated_sync
return decorated # 异步路径直接返回协程装饰器
✅ 核心优势：所有装饰逻辑（如日志、计时、参数校验、结果转换等）只需编写一次，位于 decorated 函数体内，天然消除重复。
⚠️ 重要限制与最佳实践：
- 当 func 是同步函数时，decorated 内部不可使用任何需事件循环的 asyncio 功能（如 asyncio.sleep、asyncio.create_task、await 其他协程等），否则会抛出 RuntimeError: no running event loop。
- 所有共享逻辑（Before/After 块）应严格保持同步；异步专属操作（如 await asyncio.sleep(0.1)）必须包裹在 if not is_sync: 条件内。
- 该方案兼容 functools.partial 和支持 async __call__ 的可调用对象，鲁棒性高。
使用示例：
@decorator
def sync_call() -> str:
time.sleep(0.1)
return "sync"
@decorator
async def async_call() -> str:
await asyncio.sleep(0.1) # ✅ 此处安全，因 is_sync=False
return "async"
# 混合调用
async def run():
res1 = sync_call() # → 同步执行，输出 Before/After
print(res1) # "sync"
res2 = await async_call() # → 异步执行，输出 Before/After
print(res2) # "async"
asyncio.run(run())
输出：
Before After with result='sync' sync Before After with result='async' async
此方案在 MyPy 严格模式下类型安全（在线验证），兼顾简洁性、可维护性与工程健壮性，是构建通用型跨范式装饰器的推荐范式。
Python免费学习笔记（深入）：立即使用
在学习笔记中，你将探索 Python 的核心概念和高级技巧！
