文章目录环境信息前言同样的代码有人写成一百行有人写成一行装饰器一、装饰器原理为什么一个 就能包住函数二、坑为什么装饰后函数名变了三、带参数的装饰器三层嵌套四、数据管道里的4个实用装饰器装饰器1retry —— 爬虫容错装饰器2timing —— 性能分析装饰器3cache —— API缓存装饰器4validate —— 数据校验复用第1篇的正则五、装饰器栈多个装饰器叠加六、总结环境信息项目版本/说明Python3.10标准库functools / time / functools.lru_cache依赖无需第三方库场景数据管道爬虫容错 / 性能分析 / API缓存 / 数据校验前言同样的代码有人写成一百行有人写成一行装饰器前两天我在重构自己的香港数据管道脚本。里面有一段逻辑是这样的——每次调政府API都要包一层try/except做重试每个函数开头都要start time.time()计时每段校验逻辑都要重复调用。写到第五个函数的时候我意识到我在反复写一模一样的代码只是套在不同的函数外面。这正好是装饰器的用武之地。但很多人对装饰器的理解停在就是那个 符号。这导致两个问题一是抄别人的装饰器改个参数就报错二是写不出带参数的装饰器那种三层嵌套的。这篇文章想把这两层都讲透——先讲清装饰器为什么能工作闭包原理再给4个数据管道里真实会用的装饰器。收藏提示①装饰器的本质是高阶函数——函数可以作为参数传入也可以作为返回值返回。理解了这一句话就理解了80%的装饰器。文末有4个开箱即用的装饰器。一、装饰器原理为什么一个 就能包住函数先看最朴素的装饰器——不用 语法手动包一层defmy_func(x):returnx*2# 装饰器就是一个接收函数、返回函数的高阶函数deftimer(func):importtimedefwrapper(*args,**kwargs):starttime.time()resultfunc(*args,**kwargs)print(f{func.__name__}耗时{time.time()-start:.4f}秒)returnresultreturnwrapper# 注意返回的是 wrapper不是调用结果# 手动装饰my_functimer(my_func)print(my_func(21))# 会先打印耗时再返回 42关键点就在最后一行my_func timer(my_func)。我们把my_func这个函数传给了timer然后把timer返回的新函数wrapper重新赋值给my_func。从此my_func这个变量名指向的是wrapper了。语法只是这个赋值操作的语法糖timerdefmy_func(x):returnx*2# 完全等价于# my_func timer(my_func)而wrapper能记住外层函数func也就是原来的 my_func靠的是闭包——内层函数引用了外层函数的变量即使外层函数已经返回这个变量也不会被销毁。收藏提示②装饰器 高阶函数 闭包。timer就是func timer(func)的简写。理解了这一点后面带参装饰器的三层嵌套就顺理成章了。二、坑为什么装饰后函数名变了新手最容易踩的一个坑——装饰之后函数的元信息丢了deftimer(func):importtimedefwrapper(*args,**kwargs):starttime.time()resultfunc(*args,**kwargs)print(f耗时{time.time()-start:.4f}秒)returnresultreturnwrappertimerdefmy_func(x):returnx*2print(my_func.__name__)# 输出 wrapper而不是 my_funcprint(my_func.__doc__)# None原始docstring丢了函数名变成了wrapper。这在调试、日志、序列化时会造成混乱。修复方式是用functools.wrapsfromfunctoolsimportwrapsimporttimedeftimer(func):wraps(func)# 关键保留原函数的元信息defwrapper(*args,**kwargs):starttime.time()resultfunc(*args,**kwargs)print(f耗时{time.time()-start:.4f}秒)returnresultreturnwrappertimerdefmy_func(x):这是原始文档字符串returnx*2print(my_func.__name__)# my_func ✓print(my_func.__doc__)# 这是原始文档字符串 ✓规则只要写装饰器就必须加wraps(func)。没有例外。三、带参数的装饰器三层嵌套前面讲的装饰器是两层嵌套timer 里套 wrapper。但如果我想让装饰器接受参数——比如retry(times3)而不是固定重试次数——就需要三层嵌套fromfunctoolsimportwrapsimporttimedefretry(times3,delay1):带参数的装饰器允许自定义重试次数和间隔defdecorator(func):# 第二层接收被装饰的函数wraps(func)defwrapper(*args,**kwargs):# 第三层接收函数的实际参数forattemptinrange(times):try:returnfunc(*args,**kwargs)exceptExceptionase:ifattempttimes-1:raise# 最后一次还是失败抛出print(f重试{attempt1}/{times}{e})time.sleep(delay)returnNonereturnwrapperreturndecorator# 第一层返回装饰器# 使用retry(times3,delay0.5)deffetch_data(url):# 模拟不稳定的API调用importrandomifrandom.random()0.7:raiseConnectionError(连接超时)return数据resultfetch_data(https://api.example.com)print(result)理解三层嵌套的关键是把retry(times3, delay0.5)拆开看# retry(times3, delay0.5) 等价于# fetch_data retry(times3, delay0.5)(fetch_data)# └────── 先调用 retry返回 decorator ──────┘# └── 再用 decorator 装饰 fetch_data ──┘第一层retry接收装饰器自己的参数返回第二层decorator第二层接收被装饰的函数返回第三层wrapper第三层才是真正被调用的函数。这就是带参装饰器三层嵌套的由来。收藏提示③无参装饰器两层嵌套带参装饰器三层嵌套。背住这个规律以后手写装饰器不会再懵。核心是记住deco(args)等于func deco(args)(func)。四、数据管道里的4个实用装饰器下面4个装饰器都是我在香港数据管道项目里真实用过的。它们覆盖了数据工程的四个高频需求容错、性能、缓存、校验。装饰器1retry —— 爬虫容错调用政府API时网络抖动、限流、服务器临时故障是常态。手动写try/except重试会让代码又臭又长fromfunctoolsimportwrapsimporttimedefretry(times3,delay1,exceptions(Exception,)):defdecorator(func):wraps(func)defwrapper(*args,**kwargs):forattemptinrange(1,times1):try:returnfunc(*args,**kwargs)exceptexceptionsase:ifattempttimes:print(f[retry]{func.__name__}失败{times}次放弃:{e})raiseprint(f[retry]{func.__name__}第{attempt}次失败({e}){delay}秒后重试)time.sleep(delay)returnwrapperreturndecoratorretry(times3,delay0.5)deffetch_mtr_schedule(line,station):# 调用港铁API不稳定的模拟importrandomifrandom.random()0.5:raiseConnectionError(rt.data.gov.hk 连接超时)returnf{line}{station}到站信息装饰器2timing —— 性能分析数据管道里哪个环节最慢给每个函数加计时一眼定位瓶颈fromfunctoolsimportwrapsimporttimedeftiming(func):wraps(func)defwrapper(*args,**kwargs):starttime.perf_counter()resultfunc(*args,**kwargs)elapsedtime.perf_counter()-startifelapsed1:# 只打印慢于1秒的避免日志刷屏print(f[timing]{func.__name__}耗时{elapsed:.2f}秒 ⚠️)else:print(f[timing]{func.__name__}耗时{elapsed:.3f}秒)returnresultreturnwrappertimingdefprocess_rental_csv():time.sleep(2.5)# 模拟慢操作return处理完成process_rental_csv()# 输出: [timing] process_rental_csv 耗时 2.50秒 ⚠️装饰器3cache —— API缓存同一个API短时间内重复调用纯属浪费。用一个简单的内存缓存装饰器fromfunctoolsimportwrapsimporttimedefcache(ttl30):带TTL的简单内存缓存装饰器_cache{}defdecorator(func):wraps(func)defwrapper(*args,**kwargs):keystr(args)str(sorted(kwargs.items()))ifkeyin_cache:value,timestamp_cache[key]iftime.time()-timestampttl:returnvalue resultfunc(*args,**kwargs)_cache[key](result,time.time())returnresultreturnwrapperreturndecoratorcache(ttl10)defget_air_quality(station):# 模拟昂贵的API调用time.sleep(1)returnf{station}AQHI 数据提示Python 标准库自带了functools.lru_cache如果只需要无过期时间的缓存直接用现成的不必自己写。我这里的cache(ttl30)是为了支持缓存过期这个 lru_cache 不提供的功能。装饰器4validate —— 数据校验复用第1篇的正则这是第1篇正则校验码的延续。把身份证校验逻辑封装成装饰器函数只处理已经校验通过的数据fromfunctoolsimportwrapsdefvalidate_hkid(func):校验参数中的香港身份证号码无效则跳过wraps(func)defwrapper(hkid,*args,**kwargs):# 复用第1篇的校验逻辑importre patternre.compile(r^([A-Z]{1,2})(\d{6})[(]([0-9A])[)]$)mpattern.match(hkid.strip().upper())ifnotm:print(f[validate] 跳过无效格式:{hkid})returnNone# ... 这里省略校验码算法的完整实现见第1篇returnfunc(hkid,*args,**kwargs)returnwrappervalidate_hkiddefprocess_record(hkid):returnf处理记录:{hkid}print(process_record(A123456(3)))# 处理print(process_record(INVALID))# 跳过五、装饰器栈多个装饰器叠加一个函数可以同时套多个装饰器从下往上依次执行timingretry(times3,delay0.5)cache(ttl10)deffetch_data(url):time.sleep(1)return数据执行顺序fetch_data先被cache包裹再被retry包裹最后被timing包裹。调用时的执行流程是timing → retry → cache → fetch_data返回时反向。注意装饰器的顺序如果cache在最内层最靠近函数那么缓存命中时retry和timing都不会执行——这正是我们想要的缓存命中就不需要重试和计时了。六、总结装饰器不是高级技巧而是工程化的基本功。它的本质就是一句话函数是一等公民可以传参数、可以当返回值。四个要点装饰器 高阶函数 闭包deco就是func deco(func)必须加wraps(func)否则函数名和 docstring 会丢无参两层、带参三层背住deco(args)deco(args)(func)装饰器栈从下往上执行内层装饰器如缓存能短路外层装饰器如重试这4个装饰器retry/timing/cache/validate覆盖了数据管道的四个高频需求。把它们套在你自己的函数上你会发现那些重复的try/except、time.time()、缓存字典、校验判断全都消失了。这篇和第1篇正则校验码是一个闭环——第1篇写校验逻辑怎么实现这篇写校验逻辑怎么优雅地集成到代码里。本文为 Python 装饰器技术分享。4个装饰器均为作者在数据管道项目中的实际应用。validate 装饰器复用了本账号0814第1篇文章中的香港身份证校验码算法。