装饰器是 Python 面试必考、项目必用的特性。但很多人只会写timer打印耗时遇到真实需求就卡壳。本文从实际业务出发给出 5 个可复用的装饰器实现拿来即用。前置知识functools.wraps为什么必须写pythonimport functools def my_deco(func): functools.wraps(func) # 这行不能省 def wrapper(*args, **kwargs): return func(*args, **kwargs) return wrapper不写wraps被装饰函数的__name__、__doc__会丢失影响调试和序列化。这是所有装饰器的底线下文均默认带wraps。场景一函数超时强制中断Timeout微服务或爬虫中单个函数不能拖垮整个进程。用signal或threading实现超时控制这里用signal仅 Unix 适用Windows 需改用multiprocessingpythonimport signal import functools class TimeoutError(Exception): pass def timeout(seconds): def decorator(func): functools.wraps(func) def wrapper(*args, **kwargs): def _handler(signum, frame): raise TimeoutError(f函数 {func.__name__} 执行超时 {seconds}s) old_handler signal.signal(signal.SIGALRM, _handler) signal.alarm(seconds) try: result func(*args, **kwargs) finally: signal.alarm(0) signal.signal(signal.SIGALRM, old_handler) return result return wrapper return decorator # 使用 timeout(3) def slow_query(): import time time.sleep(5) return done # slow_query() 将抛出 TimeoutError注意signal只能用于主线程异步或多线程场景请改用asyncio.wait_for。场景二带过期时间的缓存TTL Cache替代functools.lru_cache支持自定义过期时间适合接口数据缓存pythonimport time import functools from collections import OrderedDict def ttl_cache(ttl_seconds60, maxsize128): def decorator(func): cache OrderedDict() functools.wraps(func) def wrapper(*args, **kwargs): key (args, tuple(kwargs.items())) now time.time() # 如果缓存存在且未过期 if key in cache: value, timestamp cache[key] if now - timestamp ttl_seconds: # 移到末尾表示最近使用 cache.move_to_end(key) return value # 缓存失效或不存在重新计算 result func(*args, **kwargs) cache[key] (result, now) cache.move_to_end(key) if len(cache) maxsize: cache.popitem(lastFalse) # 淘汰最早 return result return wrapper return decorator ttl_cache(ttl_seconds5) def get_config(): import random return random.randint(1, 100) print(get_config()) # 第一次随机 print(get_config()) # 5秒内返回相同值场景三重试机制含指数退避调用外部 API 时网络抖动是常态。这个装饰器支持重试次数、延迟和退避因子pythonimport time import functools def retry(max_attempts3, delay1, backoff2, exceptions(Exception,)): def decorator(func): functools.wraps(func) def wrapper(*args, **kwargs): _delay delay for attempt in range(1, max_attempts 1): try: return func(*args, **kwargs) except exceptions as e: if attempt max_attempts: raise print(f重试 {attempt}/{max_attempts}等待 {_delay}s错误: {e}) time.sleep(_delay) _delay * backoff return None # 不会执行到这里 return wrapper return decorator retry(max_attempts4, delay0.5, backoff2) def unstable_request(): import random if random.random() 0.7: raise ConnectionError(网络错误) return 成功场景四权限校验基于角色Web 项目中常需要检查用户角色用装饰器分离业务逻辑pythonfrom functools import wraps from flask import request, abort # 或 django 的 request def require_role(allowed_roles): def decorator(func): wraps(func) def wrapper(*args, **kwargs): # 假设从请求头或 session 获取当前用户角色 user_role request.headers.get(X-User-Role, guest) if user_role not in allowed_roles: abort(403, description权限不足) return func(*args, **kwargs) return wrapper return decorator # 使用 require_role([admin, editor]) def delete_post(post_id): return f已删除 {post_id}场景五函数执行耗时与内存占用监控Debug 利器生产环境排查性能问题记录耗时和内存增量pythonimport time import tracemalloc import functools def monitor(func): functools.wraps(func) def wrapper(*args, **kwargs): tracemalloc.start() start time.perf_counter() result func(*args, **kwargs) elapsed time.perf_counter() - start current, peak tracemalloc.get_traced_memory() tracemalloc.stop() print(f[{func.__name__}] 耗时: {elapsed:.4f}s, 内存峰值: {peak / 1024:.2f} KB) return result return wrapper monitor def heavy_process(): data [i**2 for i in range(100000)] return sum(data)装饰器叠加顺序多个装饰器从下往上执行靠近函数的先执行pythonretry(max_attempts3) monitor def fetch_data(): pass # 执行顺序先 monitor 进入再 retry 进入退出时 retry 先退出。避坑总结坑点解决方案忘了wraps一律加上避免元信息丢失装饰器参数可变用*args, **kwargs透传类方法装饰器注意self参数wrapper 中保留*args异步函数装饰用async def wrapperawait func()缓存内存爆炸设置maxsize并淘汰策略结语以上 5 个装饰器覆盖了超时、缓存、重试、权限、监控五个高频场景全部代码已在 Python 3.10 测试通过。你可以直接复制到项目 utils 中按需调整参数。装饰器本质是“闭包 函数对象”理解透了就能写出符合业务需求的定制化工具。如果你有其它场景想用装饰器实现欢迎在评论区讨论。本文无推广、无引流纯粹技术分享希望能帮到正在写项目的你。