1.3 装饰器与上下文管理器

第一阶段 · 1.3 装饰器与上下文管理器学习目标彻底掌握 Python 中用于代码复用和资源管理的高级特性理解它们在 FastAPI 中的底层应用。预计用时2 天每天约 3 小时重要程度⭐⭐⭐⭐FastAPI 的依赖注入、路由定义、生命周期管理大量依赖这两个特性 为什么先学这个在 FastAPI 的开发中你几乎每天都会和它们打交道装饰器app.get(/)就是一个装饰器它将你的普通函数变成了 HTTP 路由。上下文管理器当你连接数据库如async with session:或者管理 FastAPI 的启动和关闭Lifespan时都会用到它。一句话总结掌握了装饰器你就能看懂 FastAPI 是怎么把路由和函数绑定的掌握了上下文管理器你就学会了如何优雅且安全地管理数据库等外部资源。 第一部分装饰器 (Decorators)1. 什么是装饰器装饰器本质上是一个函数。它的作用是在不修改原函数代码的前提下给原函数增加额外的功能比如打印日志、计算耗时、权限校验。你可以把它想象成一个包装盒你把原来的函数放进盒子里盒子外面加了点新花样然后再把盒子还给你。# ❌ 没有装饰器你需要修改原函数代码来增加计时功能importtimedefheavy_computation():start_timetime.time()print(开始计算...)time.sleep(1)print(计算结束)end_timetime.time()print(f耗时:{end_time-start_time}秒)# ✅ 有装饰器原函数保持纯粹计时功能被抽离出来importtime# 这是一个装饰器deftimer_decorator(func):defwrapper(*args,**kwargs):start_timetime.time()resultfunc(*args,**kwargs)# 执行原来的函数end_timetime.time()print(f函数{func.__name__}执行耗时:{end_time-start_time:.4f}秒)returnresultreturnwrapper# 使用 语法糖应用装饰器timer_decoratordefheavy_computation():print(开始计算...)time.sleep(1)print(计算结束)heavy_computation()# 输出:# 开始计算...# 计算结束# 函数 heavy_computation 执行耗时: 1.00xx 秒2. functools.wraps保留原函数的身份证当你使用装饰器时实际上发生的是偷梁换柱。timer_decorator相当于执行了heavy_computation timer_decorator(heavy_computation)。此时heavy_computation已经变成了内部的wrapper函数这会导致原函数的名字__name__和注释__doc__丢失。在 FastAPI 中框架依赖这些元信息来生成 Swagger API 文档所以丢失元信息是致命的。fromfunctoolsimportwrapsdefmy_decorator(func):# 使用 wraps 将原函数 (func) 的名字和注释复制给 wrapperwraps(func)defwrapper(*args,**kwargs):print(调用前执行)returnfunc(*args,**kwargs)returnwrappermy_decoratordefsay_hello():这是一个打招呼的函数print(Hello!)# 因为用了 wraps下面两行输出正常print(say_hello.__name__)# 输出: say_helloprint(say_hello.__doc__)# 输出: 这是一个打招呼的函数⚠️强制规范以后只要写装饰器必须加上wraps(func)3. 带参数的装饰器有时候我们希望装饰器本身也能接收参数。比如写一个重试装饰器我们希望指定重试次数。这就需要在装饰器外面再包一层函数被称为装饰器工厂。fromfunctoolsimportwrapsimporttime# 第一层接收装饰器参数defretry(max_retries3,delay1): 当函数抛出异常时自动重试 # 第二层真正的装饰器接收原函数defdecorator(func):wraps(func)# 第三层包装器执行实际逻辑defwrapper(*args,**kwargs):forattemptinrange(1,max_retries1):try:returnfunc(*args,**kwargs)exceptExceptionase:print(f[{func.__name__}] 第{attempt}次失败:{e})ifattemptmax_retries:raisee# 最后一次失败则向上抛出time.sleep(delay)returnwrapperreturndecorator# 使用带参数的装饰器retry(max_retries2,delay0.5)defunstable_request():print(发起网络请求...)raiseConnectionError(网络超时)# 运行 unstable_request() 会自动重试 2 次4. 类装饰器了解即可除了函数Python 的类也可以作为装饰器核心是实现__call__魔术方法。这种方式常用于需要保存状态的场景比如计数器。classCountCalls:def__init__(self,func):self.funcfunc self.num_calls0# 状态保存在实例属性中def__call__(self,*args,**kwargs):self.num_calls1print(f函数{self.func.__name__}已被调用{self.num_calls}次)returnself.func(*args,**kwargs)CountCallsdefsay_hi():print(Hi!)say_hi()# 调用 1 次say_hi()# 调用 2 次 第二部分上下文管理器 (Context Managers)1. 什么是上下文管理器核心作用优雅且安全地管理资源的获取与释放。最典型的例子是打开文件。如果你忘记关闭文件会导致内存泄漏。上下文管理器配合with关键字可以保证无论代码是否发生异常资源都会被正确释放。只要一个对象实现了__enter__()和__exit__()方法它就是一个上下文管理器。# ❌ 没有上下文管理器容易忘记关闭或者中间报错导致没执行到 closefopen(test.txt,w)f.write(hello)f.close()# ✅ 有上下文管理器即使 write 报错文件也会自动关闭classFileManager:def__init__(self,filename,mode):self.filenamefilename self.modemodedef__enter__(self):print( 打开文件)self.fileopen(self.filename,self.mode)returnself.file# 这里的返回值会赋给 as 后面的变量 fdef__exit__(self,exc_type,exc_val,exc_tb):print( 关闭文件)ifself.file:self.file.close()# 注意如果有异常这里可以决定是否吞没异常withFileManager(test.txt,w)asf:f.write(Hello Context Manager!)2. contextlib.contextmanager简写语法每次写一个类并实现两个魔术方法太麻烦了。Python 提供了contextmanager装饰器让你用一个生成器函数就能快速创建上下文管理器。fromcontextlibimportcontextmanagercontextmanagerdefopen_file(filename,mode):print( 准备打开文件)fopen(filename,mode)try:# yield 之前的部分相当于 __enter__# yield 抛出的值就是 as 后面的变量yieldffinally:# yield 之后finally 块中相当于 __exit__# 无论有没有报错finally 都会执行print( 准备关闭文件)f.close()withopen_file(test.txt,r)asf:# do something...pass在 FastAPI 中的应用FastAPI 的依赖注入系统Depends大量使用这种yield语法来管理数据库会话Session。请求开始时创建 Session请求结束时关闭 Session。3. 异步上下文管理器FastAPI 高频使用在 FastAPI 这样的异步框架中我们连接的数据库、HTTP 客户端通常都是异步的。因此我们需要异步上下文管理器也就是配合async with使用。协议方法变成了__aenter__和__aexit__。简写装饰器变成了asynccontextmanager。importasynciofromcontextlibimportasynccontextmanager# 模拟一个异步的数据库连接asynccontextmanagerasyncdefasync_db_session():print( [开始] 创建异步数据库会话...)awaitasyncio.sleep(0.5)# 模拟网络延迟try:yieldDB_Session_Objectfinally:print( [结束] 关闭异步数据库会话...)awaitasyncio.sleep(0.5)asyncdefmain():# 使用 async withasyncwithasync_db_session()assession:print(f 正在使用:{session}执行查询...)# asyncio.run(main())在 FastAPI 中的应用LifespanFastAPI 允许你定义一个在 App 启动前执行、App 关闭后执行的代码块比如初始化数据库连接池这就是通过asynccontextmanager实现的。✏️ 练习题练习 1编写一个参数类型检查装饰器编写一个装饰器check_types它会在函数执行前检查传入的参数类型是否与函数签名的类型注解一致。如果不一致抛出TypeError。# 请实现 check_types 装饰器check_typesdefmultiply(a:int,b:int)-int:returna*b# multiply(2, 3) # 应该成功# multiply(2, 3) # 应该抛出 TypeError 参考答案fromfunctoolsimportwrapsimportinspectdefcheck_types(func):wraps(func)defwrapper(*args,**kwargs):# 获取函数的签名信息siginspect.signature(func)# 将传入的参数与签名绑定bound_argssig.bind(*args,**kwargs)bound_args.apply_defaults()forname,valueinbound_args.arguments.items():# 获取该参数的期望类型expected_typesig.parameters[name].annotationifexpected_typeisnotinspect._empty:ifnotisinstance(value,expected_type):raiseTypeError(f参数 {name} 期望类型{expected_type}但收到了{type(value)})returnfunc(*args,**kwargs)returnwrapper练习 2实现一个计时的异步上下文管理器编写一个类AsyncTimer配合async with使用在退出时打印代码块的执行耗时。 参考答案importtimeimportasyncioclassAsyncTimer:asyncdef__aenter__(self):self.starttime.time()print(异步计时开始...)returnselfasyncdef__aexit__(self,exc_type,exc_val,exc_tb):costtime.time()-self.startprint(f异步计时结束耗时:{cost:.4f}秒)# 测试# async def test():# async with AsyncTimer():# await asyncio.sleep(1) 关键概念总结卡片┌──────────────────────────────────────────────────────┐ │ 装饰器 (Decorators) 速查 │ ├──────────────────────────────────────────────────────┤ │ 本质接收函数并返回新函数的高阶函数。 │ │ 作用不改原代码增加新功能日志、鉴权、路由。 │ │ 强制规范必须在 wrapper 上加 wraps(func) 保留元信息。 │ │ 带参数装饰器需要三层嵌套工厂 - decorator - wrapper。 │ ├──────────────────────────────────────────────────────┤ │ 上下文管理器 (Context Managers) 速查 │ ├──────────────────────────────────────────────────────┤ │ 本质实现 __enter__ 和 __exit__ 的对象。 │ │ 作用配合 with 语句保证资源文件、连接必定被安全释放。 │ │ 快捷写法contextmanager yield 生成器。 │ │ 异步场景__aenter__ / __aexit__配合 async with。 │ │ 异步快捷asynccontextmanager async def yield。 │ └──────────────────────────────────────────────────────┘ 下一步学完本节后你已经补齐了所有的 Python 前置基础接下来你将正式进入第二阶段Web 后端基础知识。