本文介绍一种巧妙利用协程 send() 方法特性的装饰器实现方案，避免在同步/异步分支中重复编写“前置处理→调用原函数→后置处理”逻辑，真正实现单份核心逻辑、双模式无缝适配。
本文介绍一种巧妙利用协程 send() 方法特性的装饰器实现方案，避免在同步/异步分支中重复编写“前置处理→调用原函数→后置处理”逻辑，真正实现单份核心逻辑、双模式无缝适配。
在 Python 中为同步（def）和异步（async def）函数编写统一装饰器时，常见做法是通过 asyncio.iscoroutinefunction() 分支判断，再分别定义 async def decorated(...) 和 def decorated(...) —— 但这导致前置/后置逻辑被完全复制两次，违背 DRY 原则，也增加维护成本。
幸运的是，Python 协程对象具有可迭代性：一个 async def 函数返回的协程对象支持 .send(None) 触发执行，并在首次 await 或协程结束时抛出 StopIteration 异常，其 value 属性即为返回值。这一特性使我们能以同步方式“驱动”协程完成，无需事件循环参与 —— 从而将异步装饰器逻辑“复用”于同步场景。
以下是无代码重复的通用装饰器实现：
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]]]:
"""精确判断是否为协程函数（支持 partial 和 __call__ 类）"""
while isinstance(func, functools.partial):
func = func.func
if asyncio.iscoroutinefunction(func):
return True
if hasattr(func, "__call__"):
return asyncio.iscoroutinefunction(func.__call__)
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("Unexpected coroutine state") # mypy 友好兜底
return decorated_sync
return decorated # 异步路径直接返回协程装饰器
✅ 关键优势与使用说明
零逻辑重复：Before / After 等共用逻辑仅在 async def decorated 中定义一次；
类型安全：借助 TypeGuard 和 ParamSpec + TypeVar，完整保留原函数签名与泛型类型；
兼容类方法：is_coroutine 支持检测含 async def __call__ 的可调用类；
-
开箱即用：
@decorator def sync_call() -> str: time.sleep(0.1) return "sync" @decorator async def async_call() -> str: await asyncio.sleep(0.1) return "async" # 调用方式与原函数一致 print(sync_call()) # → "sync" print(asyncio.run(async_call())) # → "async"
⚠️ 重要注意事项
- 禁止在同步路径中使用 asyncio 功能：当 is_sync=True 时，协程在无事件循环环境下运行，所有依赖 asyncio.get_event_loop() 或 await 的代码（如 asyncio.sleep, aiohttp）将崩溃；
- 安全实践：所有共享逻辑（Before/After）应保持同步；异步操作（如日志异步写入、DB 查询）必须包裹在 if not is_sync: 分支内；
- 调试友好：print(f"{is_sync=}") 可帮助快速验证模式识别是否正确。
该方案本质是将“装饰逻辑”升维至协程层面，再通过运行时驱动机制桥接同步世界，是 Python 协程底层机制的一次优雅应用。它不仅解决重复代码问题，更体现了对语言特性的深度理解与工程化抽象能力。
Python免费学习笔记（深入）：立即使用
在学习笔记中，你将探索 Python 的核心概念和高级技巧！
