当装饰器遇上 async如何写出同时兼容同步与异步的 Python 装饰器以统一打点 SDK 为例从 Flask 到 FastAPI一器两用一、问题从何而来一个常见但容易被低估的工程场景你的团队维护一套统一监控打点 SDK需要在函数执行前后记录耗时、埋入 trace_id、上报异常。早期服务跑在Flask同步 WSGI后来新项目迁移到FastAPI异步 ASGI。于是问题来了——# 同步环境Flaskmonitordeffetch_user(user_id):returndb.query(User).get(user_id)# 异步环境FastAPImonitorasyncdeffetch_user(user_id):returnawaitdb.fetch_user(user_id)同一个monitor能同时正确地装饰这两类函数吗直觉上可以实际上会翻车。普通装饰器拿到 async 函数后返回值会变成一个协程对象而非执行结果导致下游调用链断裂。反过来用asyncio.iscoroutinefunction手动分支虽然可行但写法冗余、容易遗漏。下面我们从原理出发逐步构建一个健壮的双模装饰器。二、先看清对手同步与异步的本质差异在动手写代码之前有必要把两种调用模型捋清楚。维度同步函数异步函数关键字defasync def调用方式result func()result await func()返回值直接结果协程对象coroutine阻塞行为直接阻塞线程让出控制权给事件循环检测方式asyncio.iscoroutinefunction(func)同左返回True关键矛盾装饰器本质是wrapper(*args, **kwargs)而wrapper本身的同步/异步属性决定了它能否正确调度被装饰函数。如果wrapper是同步的✅ 装饰同步函数 → 正常❌ 装饰异步函数 → 返回 coroutine不会真正执行如果wrapper是异步的❌ 装饰同步函数 → 调用方不会await报错✅ 装饰异步函数 → 正常所以核心思路是运行时判断被装饰函数的类型动态选择 wrapper 的同步/异步版本。三、第一版判断分流——最朴素的双模方案先给出一个直觉式的实现虽然能跑但有明显的代码味问题importtimeimportasyncioimportfunctoolsimportlogging loggerlogging.getLogger(__name__)defmonitor_v1(func):第一版if/else 分流ifasyncio.iscoroutinefunction(func):functools.wraps(func)asyncdefasync_wrapper(*args,**kwargs):starttime.perf_counter()try:resultawaitfunc(*args,**kwargs)elapsedtime.perf_counter()-start logger.info(f[ASYNC]{func.__name__}耗时{elapsed:.4f}s)returnresultexceptExceptionase:elapsedtime.perf_counter()-start logger.error(f[ASYNC]{func.__name__}异常{e}耗时{elapsed:.4f}s)raisereturnasync_wrapperelse:functools.wraps(func)defsync_wrapper(*args,**kwargs):starttime.perf_counter()try:resultfunc(*args,**kwargs)elapsedtime.perf_counter()-start logger.info(f[SYNC]{func.__name__}耗时{elapsed:.4f}s)returnresultexceptExceptionase:elapsedtime.perf_counter()-start logger.error(f[SYNC]{func.__name__}异常{e}耗时{elapsed:.4f}s)raisereturnsync_wrapper能用但痛点明显同步/异步的逻辑几乎完全重复只是await的有无一旦打点逻辑变复杂加 trace_id、上报 metric两处都要改违反 DRY 原则四、第二版利用inspect与asyncio优雅合并核心改进思路把重复的计时 异常捕获逻辑抽成独立函数wrapper 只负责调度。importtimeimportasyncioimportfunctoolsimportinspectimportloggingfromtypingimportCallable,TypeVar,ParamSpec loggerlogging.getLogger(__name__)PParamSpec(P)TTypeVar(T)defmonitor(func:Callable[P,T])-Callable[P,T]: 双模装饰器同时支持同步函数和异步函数。 设计原则 1. 运行时判断被装饰函数类型 2. 通用逻辑提取避免代码重复 3. 完整保留原函数签名functools.wraps def_log_execution(func_name:str,elapsed:float,error:ExceptionNone):统一的打点逻辑同步/异步共用iferror:logger.error(f⛔{func_name}异常{error}耗时{elapsed:.4f}s)else:logger.info(f✅{func_name}执行完成耗时{elapsed:.4f}s)ifasyncio.iscoroutinefunction(func):functools.wraps(func)asyncdefasync_wrapper(*args,**kwargs):starttime.perf_counter()try:resultawaitfunc(*args,**kwargs)returnresultexceptExceptionase:_log_execution(func.__name__,time.perf_counter()-start,e)raiseelse:_log_execution(func.__name__,time.perf_counter()-start)returnasync_wrapperelse:functools.wraps(func)defsync_wrapper(*args,**kwargs):starttime.perf_counter()try:resultfunc(*args,**kwargs)returnresultexceptExceptionase:_log_execution(func.__name__,time.perf_counter()-start,e)raiseelse:_log_execution(func.__name__,time.perf_counter()-start)returnsync_wrapper比第一版好一些但async_wrapper和sync_wrapper里仍然有try/except/else的重复结构。五、第三版终极方案——带参数的双模装饰器工厂实际 SDK 场景中装饰器往往需要参数如上报的 service 名、采样率等。我们一并解决 monitor_sdk.py — 统一打点 SDK 核心模块 同时兼容 Flask同步 WSGI和 FastAPI异步 ASGI importtimeimportasyncioimportfunctoolsimportinspectimportloggingimportuuidfromtypingimportCallable,Any,Optional,TypeVar,ParamSpec loggerlogging.getLogger(monitor_sdk)PParamSpec(P)TTypeVar(T)classExecutionContext:执行上下文封装一次调用的全部打点信息def__init__(self,func_name:str,service:str):self.func_namefunc_name self.serviceservice self.trace_idstr(uuid.uuid4())[:8]self.start_time:float0self.elapsed:float0self.success:boolTrueself.error:Optional[Exception]Nonedefstart(self):self.start_timetime.perf_counter()deffinish(self,error:ExceptionNone):self.elapsedtime.perf_counter()-self.start_timeiferror:self.successFalseself.errorerrordefreport(self):统一上报逻辑示例写日志实际可替换为 Prometheus、StatsD 等status✅ 成功ifself.successelsef⛔ 失败({type(self.error).__name__})logger.info(f[{self.service}]{self.func_name}| ftrace_id{self.trace_id}| fstatus{status}| felapsed{self.elapsed:.4f}s)defmonitor(_func:Optional[Callable]None,*,service:strdefault,log_args:boolFalse,): 双模打点装饰器 —— 同时支持同步与异步函数。 参数 _func: 被装饰的函数支持 monitor 和 monitor(servicexx) 两种语法 service: 服务标识用于区分不同微服务的打点 log_args: 是否记录函数入参调试时开启 使用方式 monitor # 最简用法 monitor(serviceuser-service) # 带参数 monitor(servicepay, log_argsTrue) # 记录入参 defdecorator(func:Callable[P,T])-Callable[P,T]:is_asyncasyncio.iscoroutinefunction(func)modeasyncifis_asyncelsesyncdef_build_context(*args,**kwargs)-ExecutionContext:构建执行上下文ctxExecutionContext(func.__name__,service)iflog_args:logger.debug(f[{service}]{func.__name__}调用参数: fargs{args}, kwargs{kwargs})returnctxifis_async:functools.wraps(func)asyncdefasync_wrapper(*args:P.args,**kwargs:P.kwargs)-T:ctx_build_context(*args,**kwargs)ctx.start()try:resultawaitfunc(*args,**kwargs)returnresultexceptExceptionase:ctx.finish(errore)raiseelse:ctx.finish()finally:ctx.report()returnasync_wrapperelse:functools.wraps(func)defsync_wrapper(*args:P.args,**kwargs:P.kwargs)-T:ctx_build_context(*args,**kwargs)ctx.start()try:resultfunc(*args,**kwargs)returnresultexceptExceptionase:ctx.finish(errore)raiseelse:ctx.finish()finally:ctx.report()returnsync_wrapper# 支持 monitor 和 monitor() 两种调用方式if_funcisnotNone:returndecorator(_func)returndecorator六、完整实战Flask FastAPI 统一接入6.1 Flask 同步服务# app_flask.pyfromflaskimportFlask,jsonifyfrommonitor_sdkimportmonitor appFlask(__name__)app.route(/user/int:user_id)monitor(serviceuser-service,log_argsTrue)defget_user(user_id:int):模拟同步数据库查询importtime time.sleep(0.1)# 模拟 I/Oreturnjsonify({user_id:user_id,name:张三})app.route(/order/int:order_id)monitor(serviceorder-service)defget_order(order_id:int):模拟同步逻辑异常iforder_id0:raiseValueError(订单ID不能为负数)returnjsonify({order_id:order_id,status:shipped})if__name____main__:app.run(port5000)6.2 FastAPI 异步服务# app_fastapi.pyimportasynciofromfastapiimportFastAPI,HTTPExceptionfrommonitor_sdkimportmonitor appFastAPI()app.get(/user/{user_id})monitor(serviceuser-service,log_argsTrue)asyncdefget_user(user_id:int):模拟异步数据库查询awaitasyncio.sleep(0.1)# 模拟异步 I/Oreturn{user_id:user_id,name:李四}app.get(/order/{order_id})monitor(serviceorder-service)asyncdefget_order(order_id:int):模拟异步逻辑异常iforder_id0:raiseValueError(订单ID不能为负数)awaitasyncio.sleep(0.05)return{order_id:order_id,status:processing}# 同一个项目里也可以有同步端点app.get(/health)monitor(serviceinfra)defhealth_check():同步健康检查——同一个装饰器零改动return{status:ok}注意health_check是同步函数但 FastAPI 允许同步端点会在线程池中执行我们的monitor自动识别并适配。七、关键陷阱与避坑指南❌ 陷阱一忘记functools.wraps# 错误示范defmonitor(func):asyncdefwrapper(*args,**kwargs):# 没有 functools.wraps...returnwrapper后果func.__name__变成wrapper路由注册失败Flask/FastAPI 依赖函数名区分路由调试时日志全是wrapper。❌ 陷阱二在同步 wrapper 中await# 编译错误defsync_wrapper(*args,**kwargs):resultawaitfunc(*args,**kwargs)# SyntaxError!这就是为什么必须在定义时就确定 wrapper 的同步/异步类型而不是运行时混用。❌ 陷阱三inspect.iscoroutinefunctionvsasyncio.iscoroutinefunction两者对大部分情况一致但存在细微差异importinspectimportasyncioasyncdeff():passprint(inspect.iscoroutinefunction(f))# Trueprint(asyncio.iscoroutinefunction(f))# True推荐使用asyncio.iscoroutinefunction因为它是asyncio模块的官方检测方式与事件循环的语义更一致。注意对于functools.partial包装过的函数两者都可能返回False需要额外处理。❌ 陷阱四装饰器叠加顺序# 错误顺序app.get 在外层时monitor 接收的是路由返回值不是函数app.get(/user/{user_id})monitor(serviceuser-service)asyncdefget_user(user_id:int):...实际上面的顺序是正确的——monitor先执行装饰函数后传给app.get。关键是monitor不能出现在框架装饰器之上。✅ 避坑小结陷阱解决方案函数签名丢失始终使用functools.wraps(func)同步/异步混用定义时分支不运行时判断partial 函数检测失败用inspect.unwrap()递归解包装饰器顺序框架路由装饰器在最外层异常不应被吞try/finally中report异常原样raise八、进阶让装饰器支持类方法和静态方法实际 SDK 中装饰器可能装饰类方法。需要额外注意self参数的处理classUserService:monitor(serviceuser-service)asyncdefget_user(self,user_id:int):awaitasyncio.sleep(0.1)return{user_id:user_id}上述第三版代码已经兼容——因为*args会自然捕获self无需额外处理。唯一需要注意的是log_argsTrue时日志中会包含self对象可能产生大量无用输出。可以在_build_context中过滤def_build_context(*args,**kwargs)-ExecutionContext:ctxExecutionContext(func.__name__,service)iflog_args:# 过滤掉 self/cls 参数filtered_args[aforainargsifnot(hasattr(a,__class__)andhasattr(a,__dict__))]logger.debug(f调用参数: args{filtered_args}, kwargs{kwargs})returnctx九、性能思考分支判断的开销有人担心asyncio.iscoroutinefunction的判断有运行时开销。实际上这个判断发生在装饰时decorator 应用时而非每次调用时monitor(servicesvc)asyncdeffoo():pass# ↑ 此时 asyncio.iscoroutinefunction(foo) 已经执行完毕# ↓ 后续每次调用 foo() 走的都是固定的 async_wrapper零分支开销这正是装饰器模式的优势一次判断永久绑定。比每次调用时if inspect.iscoroutine(result)的方案高效得多。十、总结回顾整个设计思路问题本质同步/异步函数的调用协议不同wrapper 必须匹配核心策略装饰时而非运行时判断函数类型选择对应 wrapper代码复用通过ExecutionContext类抽取公共逻辑避免双倍维护工程适配支持monitor和monitor()两种语法兼容类方法、静态方法避坑要点functools.wraps、装饰器顺序、异常不吞、partial 兼容一句话总结不要试图写一个既能 await 又能不 await的 wrapper——而是在装饰时就决定好它的命运。这既是 Python 类型系统的约束也是它的优雅之处。最后分享一个实际项目中的经验统一打点 SDK 的价值不仅在于监控本身更在于它强制团队建立一致的可观测性习惯。当同一个monitor能无缝跑在 Flask、FastAPI、甚至命令行脚本中时你获得的不只是一行装饰器代码而是一套贯穿整个技术栈的标准化实践。如果你在实际接入中遇到functools.partial兼容性、与contextvars联动传递 trace_id、或装饰器与pydantic验证冲突等问题欢迎在评论区交流——这些话题每一个都值得单独展开。