做 Web 开发三年我最怕的不是需求变更而是同一个项目写到第 6 个月后代码开始“发臭”路由函数越堆越长ORM 查询到处都是 N1并发一上来库存就超卖。很多人把锅甩给“架构”但我后来发现FastAPI 和 Django 里其实藏着一批不出名但异常好用的 Python 骚操作你不需要重写框架只需要换几种写法就能让项目变得既优雅又强悍。这篇东西不是什么入门教程而是我在几个实际项目里压箱底的隐藏技巧集合适合已经写过 FastAPI/Django 的朋友拿来即用。每一节我都会拆开讲原理、写法和踩坑点保证不是那种复制粘贴就能糊弄过去的八股文。1. 依赖注入不只用来拿参数FastAPI 的 Depends 高级用法很多人用 FastAPI 的Depends只做两件事拿数据库会话、拿当前登录用户。这个用法没错但太浪费了。FastAPI 的依赖注入实际上是一个完整的“依赖图”系统它允许你定义一个“返回依赖的函数”。也就是说Depends的参数不必非得是固定的函数它可以是一个函数工厂请求进来时临时生成一个专属校验器。1.1 带参数的依赖工厂同一套代码管理不同权限假设你要做权限控制最常见的想法是写一个require_admin、一个require_editor、一个require_viewer。三个函数逻辑几乎一样只是权限字符串不同代码重复不说以后加权限还要到处复制。用依赖工厂就可以把“权限字符串”变成参数from typing import Annotated from fastapi import Depends, HTTPException, status def require_permission(permission: str): def checker( user: Annotated[User, Depends(get_current_user)] ) - User: if permission not in user.permissions: raise HTTPException( status_codestatus.HTTP_403_FORBIDDEN, detailInsufficient permissions ) return user return checker app.get(/admin) def admin_dashboard( current_user: Annotated[User, Depends(require_permission(admin))] ): return {msg: fHello {current_user.name}} app.put(/posts/{post_id}) def edit_post( current_user: Annotated[User, Depends(require_permission(editor))] ): ...这里的关键点是require_permission内部返回的checker函数自己还可以声明依赖get_current_user。FastAPI 会解析整条依赖链并且checker里拿到的用户对象就是真正的当前用户。我从这套写法里吃到最大的甜头不是少写几个函数而是权限规则收敛到了一个地方以后想加“超级管理员”权限不用再写新依赖直接require_permission(super_admin)路由签名上也能直观看到权限要求。1.2 子依赖与依赖缓存Depends是可以嵌套的而且 FastAPI 默认会在同一个请求里缓存依赖的解析结果。这意味着哪怕同一个依赖被十个下游同时声明实际也只执行一次。async def get_current_user(token: Annotated[str, Depends(oauth2_scheme)]): user await cache_or_db_get_user(token) return user async def get_user_team(user: Annotated[User, Depends(get_current_user)]): return await get_team_of_user(user.id) app.get(/me/summary) async def get_summary( user: Annotated[User, Depends(get_current_user)], team: Annotated[Team, Depends(get_user_team)], ): ...在这个例子里get_user_team依赖get_current_user路由又显式依赖get_current_user因为 FastAPI 的use_cache参数默认是True所以get_current_user在整个请求生命周期里只会被调用一次。这是一笔很划算的账十个下游依赖各自声明“我要用户”数据库只查一次。如果你确实希望每次调用都重新执行依赖可以设置Depends(..., use_cacheFalse)。不过我用 FastAPI 两年真正需要关闭缓存的情况少之又少大部分时候默认行为反而帮我省掉了大量重复查询。1.3 给依赖套上 lru_cache把只读查询变成内存加速还有一种更极致的缓存玩法。如果依赖函数是完全不依赖请求上下文的纯函数比如读配置项、查固定的字典表可以直接用functools.lru_cache做进程内缓存from functools import lru_cache lru_cache(maxsize128) def get_config_by_key(key: str) - str: return db.query(Config).filter(Config.key key).first().value之后路由里写Depends(get_config_by_key(site_name))这个值在每个进程里最多查一次库。这里有几个必须记住的坑lru_cache的缓存 key 是参数所以参数必须在业务上不可变不要对依赖请求上下文的函数使用lru_cache否则request、session会被长期持有内存迟早爆掉如果需要按时间过期换cachetools.TTLCache或直接查项目里已有的 Redis总之不要硬套这个装饰器。提示这一节所有技巧的共同点是让 FastAPI 替你管理依赖关系和生命周期。依赖负责拿数据和校验路由只负责组织业务你的代码自然就清爽了。2. Pydantic v2 校验器的黑科技把脏活累活拦截在业务代码之外Python Web 项目逃不开数据校验。FastAPI 天生跟 Pydantic 深度绑定Django 里也有各种 validator但很多人只把 Pydantic 用在“声明字段类型”这一层。Pydantic v2 在Annotated和校验器上提供了非常多隐藏能力足够你把数据清洗、格式转换、跨字段校验全部放到接口层业务代码里拿到的都是“已经可信的数据”。2.1 Annotated 类型别名让路由签名瘦成一道闪电先看一段常见的路由签名app.get(/items) def list_items( page: int Query(1, ge1, le1000), size: int Query(20, ge1, le100), ): ...如果多个接口都要分页参数每个路由都要重复写一遍ge1, le1000。用Annotated做类型别名可以把约束声明一次到处复用from typing import Annotated, Literal from fastapi import Query PageNumber Annotated[int, Query(ge1, le1000)] PageSize Annotated[int, Query(ge1, le100)] SortOrder Annotated[Literal[asc, desc], Query()] app.get(/items) def list_items( page: PageNumber 1, size: PageSize 20, order: SortOrder asc, ): ...好处是显而易见的分页规则改一处所有接口同步生效。我特别想强调Annotated不是 FastAPI 特有的语法它是 Python 3.9 加入typing的标准能力FastAPI 只是识别了里面挂载的Query、Path、Depends这些元数据。你完全可以在自己的业务函数里用Annotated做参数标注比如“这个参数必须经过脱敏处理”让类型系统帮你记住约定。2.2 BeforeValidator 与 AfterValidator数据还没定型时先动手Pydantic v2 的BeforeValidator允许你在 Pydantic 做类型转换之前先对原始值做一次处理。这非常适合“容忍脏数据”的接口。举个例子前端可能传12%这种带百分号的税率后端想存小数0.12from typing import Annotated from pydantic import BaseModel, BeforeValidator def parse_percent(value) - float: if isinstance(value, str) and value.endswith(%): return float(value[:-1]) / 100 return float(value) PercentFloat Annotated[float, BeforeValidator(parse_percent)] class TaxRule(BaseModel): rate: PercentFloat请求体传{rate: 5%}Pydantic 会先调parse_percent拿到0.05再去校验类型。如果传了abcfloat(abc)抛ValueErrorFastAPI 自动转成 422。AfterValidator则是在类型转换完之后再执行适合“清洗已转换的数据”比如去掉字符串首尾空格、压缩多个空格。单个看这些功能很碎组合起来就是输入层的堡垒任何格式不干净的数据在进业务代码之前就已经被纠正了。2.3 model_validator跨字段的联动校验field_validator只能看一个字段跨字段逻辑就得用model_validator。最典型的场景是注册时的密码确认from pydantic import BaseModel, model_validator class RegisterForm(BaseModel): username: str password: str confirm_password: str model_validator(modeafter) def passwords_match(self): if self.password ! self.confirm_password: raise ValueError(password and confirm_password do not match) return selfmodeafter表示所有字段校验完成后才执行这段代码所以可以放心读取self.password和self.confirm_password。这种写法比在视图函数里手动if form.password ! form.confirm_password干净得多而且错误能统一走 Pydantic 的校验错误格式返回前端解析逻辑完全一致。2.4 ConfigDict 里的隐藏开关Pydantic v2 里ConfigDict有很多“开关”我项目里最常用到的是这几个配置项效果典型场景str_strip_whitespaceTrue所有字符串字段自动去掉首尾空格表单、用户名extraforbid请求体里出现未声明字段直接报错防止接口参数悄悄变化validate_assignmentTrue实例属性被赋值时重新校验后端修改模型字段时的保险validate_defaultTrue字段默认值也参与校验带默认值的配置类看一个具体用法from pydantic import BaseModel, ConfigDict class CreateUser(BaseModel): model_config ConfigDict( str_strip_whitespaceTrue, extraforbid, ) username: str email: str如果请求体传{username: foo , email: ab.com, role: admin}在extraforbid下会直接报 422而不是静默忽略掉role字段。对接口兼容性来说这个保护非常关键至少在字段悄悄变化时你能第一时间发现。2.5 response_model_exclude_unset响应体别泄露默认值再分享一个响应层面的技巧。假设你的ItemOut模型里discount字段默认是0.0但某个创建接口根本没设置它。如果直接返回整个模型前端就会看到discount: 0.0这会误导调用方。加上response_model_exclude_unsetTrueFastAPI 只会返回显式传过或显式赋值的字段app.post(/items, response_modelItemOut, response_model_exclude_unsetTrue) async def create_item(item: ItemIn): return ItemOut(**item.model_dump(), discount0.0)这个参数在做“部分更新”的 PATCH 接口时特别实用。以前我为了隐藏默认值手写过一堆 dict 过滤代码后来发现 FastAPI 天生就有这个能力直接省掉十几行重复劳动。3. Django ORM 的隐藏加速键Prefetch、F() 与批量更新的组合拳如果说 FastAPI 的优雅来自“依赖与类型系统”那 Django 的强悍就体现在 ORM 的细节里。我见过太多 Django 项目跑慢了不是数据库不行而是代码里全是 N1 查询和逐条 UPDATE。下面这几个技巧全是我优化了多个后台项目后沉淀下来的核心思路只有一句话能一条 SQL 解决的事情绝不用一百条。3.1 Prefetch 对象按条件预取而不是全量预取很多开发者知道prefetch_related可以一次查出关联对象但默认是全量预取。如果每个作者的帖子有几千条预取本身也会变成灾难。Prefetch对象允许你定制预取的 QuerySet 和属性名from django.db.models import Prefetch recent_posts Post.objects.filter( is_publishedTrue ).order_by(-created_at)[:3] authors Author.objects.prefetch_related( Prefetch(posts, querysetrecent_posts, to_attrrecent_posts) ).all() for author in authors: for post in author.recent_posts: print(author.name, post.title)这里to_attrrecent_posts把预取结果挂到对象上名字可以随便取关键是只有最近 3 篇会被加载。注意这个预取结果是 Python list不是 QuerySet不能再继续.filter()。如果后续要过滤要么换字段名要么在内存循环里处理。还有一个坑Prefetch的queryset不能带聚合和复杂的注解在部分数据库后端会报错或静默返回错误数据我建议先用真实的数据库方言跑一遍再上线。3.2 bulk_create 与 bulk_update批量操作的正确姿势导入一万条数据时很多人还是for obj in objs: obj.save()。这会生成一万次 SQL 往返而bulk_create常常一条 SQL 就能完成products [ Product(namefsku-{i}, pricei * 10) for i in range(10_000) ] Product.objects.bulk_create(products, batch_size500)批量更新则略有不同得先把对象加载进内存、改好字段再一次提交products Product.objects.filter(price__lt100) for p in products: p.price 10 Product.objects.bulk_update(products, [price], batch_size500)这里最容易被忽略的坑是bulk_create和bulk_update都不会触发模型的save()方法所以post_save信号、自定义逻辑全部失效。如果你依赖信号去同步搜索索引或缓存批量操作会让你的“钩子”悄悄失灵。我的习惯是在批量操作之后手动执行一次类似refresh_search_index()的清理函数把需要同步的数据显式刷新一遍。3.3 F() 表达式让 UPDATE 在数据库端原子完成计数器加一这个操作最反面的写法是article Article.objects.get(pkpk) article.views 1 article.save()这是典型的竞态条件两个请求同时读到views10各自加一写回最终变成 11 而不是 12。正确姿势是用F()表达式from django.db.models import F Article.objects.filter(pkpk).update(viewsF(views) 1)F()会在数据库端完成SET views views 1读取和更新在同一个 SQL 语句里完成不会因为 Python 层的读取延迟导致少算。类似的扣减库存可以用F(stock) - quantity更新时间戳可以直接用F(last_seen)。再进阶一点还可以配合Case、When做条件式原子更新from django.db.models import Case, When, Value Article.objects.filter(pkpk).update( statusCase( When(views__gte1000, thenValue(hot)), defaultValue(normal), ) )这一招把“先查后改”变成了“一把 UPDATE 搞定”既省往返又避免并发脏读是我做统计面板时最常用的一招。3.4 annotate Window分组排名不再需要子查询“每个分类下热度前三的帖子”这种需求以前要么三次子查询要么把数据全查回 Python 里排序。Django 的Window函数配合RowNumber可以一步搞定from django.db.models import Window, F from django.db.models.functions import RowNumber top_posts Post.objects.annotate( rnWindow( expressionRowNumber(), partition_byF(category), order_byF(views).desc(), ) ).filter(rn__lte3)partition_by相当于给数据分组order_by决定组内排序RowNumber()打出组内的行号外面再套一层filter(rn__lte3)就是每组前三。这里有个前置条件窗口函数需要数据库支持。PostgreSQL 和 MySQL 8.0 以上都没问题SQLite 3.25 之后才支持老版本 SQLite 会直接报错所以测试环境最好用真实目标数据库跑一遍不要只依赖本地 SQLite。4. 事务与并发的优雅博弈锁、提交后回调与异步限流Web 项目做到后面真正让人头疼的往往不是增删改查而是多个请求同时改同一个资源的问题。这一节集中聊聊我在项目里处理并发时的几个固定姿势。4.1 select_for_update库存扣减不再超卖经典的秒杀/下单场景库存只有 10 件100 个人同时抢。如果先查库存再减库存必然超卖。select_for_update可以在数据库层面把选中的行锁住直到事务结束from django.db import transaction def deduct_stock(product_id: int, quantity: int): with transaction.atomic(): product Product.objects.select_for_update().get(pkproduct_id) if product.stock quantity: raise ValueError(库存不足) product.stock - quantity product.save()几个关键点必须记牢。第一select_for_update必须在事务内所以整体要包在transaction.atomic()里。第二锁是有代价的事务里只做必要的查询和更新绝对不要在里面调外部 API 或做耗时的计算否则锁持有时间太长并发能力直线下降。第三如果不想一直等被锁的行可以加nowaitTrue拿不到锁直接抛异常由上层重试。还有一个容易踩的坑如果你在事务里先普通get了一次再select_for_update().get第一次查出来的对象不带锁第二次才真正锁定。正确做法是一开始就写select_for_update().get()不要多此一举。4.2 transaction.on_commit信号别半路跑飞Django 的信号post_save等默认是在事务执行过程中触发的。这带来一个很隐蔽的坑事务里发了post_save紧接着事务回滚监听函数拿到的 instance 其实并没有真正入库。比如注册用户时发欢迎邮件邮件发出去了用户创建事务却回滚了结果用户不存在邮件却发给了不存在的账号。transaction.on_commit就是专门解决这个问题的它把回调推迟到事务真正提交之后from django.db import transaction def user_post_save(sender, instance, created, **kwargs): if created: transaction.on_commit( lambda: send_welcome_email(instance.id) ) post_save.connect(user_post_save, senderUser)这样只有User.objects.create()所在的transaction.atomic()成功提交邮件才会发送。我在 FastAPI 项目里也沿用了这个思路把“发消息、清缓存、更新搜索引擎”这类副作用放到数据库事务提交成功之后再执行。这个模式能避免非常多“数据没存上但副作用已经发生”的诡异问题。4.3 asyncio.gather 与 Semaphore既快又不被打挂FastAPI 是异步框架很多人喜欢用asyncio.gather并发请求外部 API。如果没有控制并发量1000 个 URL 同时打出去下游服务很可能直接 503。Semaphore可以给并发套上一个“阀门”import asyncio import httpx async def fetch_one(client: httpx.AsyncClient, url: str, limiter: asyncio.Semaphore): async with limiter: resp await client.get(url) resp.raise_for_status() return resp.json() async def fetch_many(urls: list[str]): limiter asyncio.Semaphore(10) async with httpx.AsyncClient(timeout10) as client: tasks [fetch_one(client, url, limiter) for url in urls] results await asyncio.gather(*tasks, return_exceptionsTrue) return results这里有几个细节。limiter在fetch_many内部创建避免了 Python 3.10 以下版本跨事件循环使用Semaphore的兼容问题。return_exceptionsTrue也很重要某个请求失败不会中断其他请求你可以拿到异常结果后统一做重试或记录。关于并发上限没有万能答案我一般先压测下游接口找到它稳定响应的 TPS然后把Semaphore初始值设为 TPS 的一半观察延迟曲线再调整。先小后大、逐步调优才是稳妥路径。5. 项目组织的魔法生命周期、中间件与自定义 Manager前面讲的都是“单点技巧”但一个项目要长期可维护还需要结构层的“魔法”。这个部分聊的是代码组织怎么把全局资源、公共逻辑、业务查询收拢到合适的位置。5.1 lifespan用异步上下文管理器的思路管理 FastAPI 生命周期很多老项目还在用app.on_event(startup)初始化全局资源这个写法在新版 FastAPI 里已经不推荐了。更好的方式是lifespan它把启动和关闭逻辑收拢到一个上下文管理器里from contextlib import asynccontextmanager from fastapi import FastAPI asynccontextmanager async def lifespan(app: FastAPI): app.state.pool await create_connection_pool() app.state.cache await create_cache_client() yield await app.state.pool.close() await app.state.cache.close() app FastAPI(lifespanlifespan)启动时创建连接池、预热缓存结束时优雅回收一目了然。这比on_event强在你想让哪个资源什么时候创建、什么时候释放全都写在同一个函数里逻辑不会散落在多个装饰器之间。尤其配合app.state存全局对象路由里用request.app.state.pool取用依赖关系非常清晰。5.2 中间件BaseHTTPMiddleware 之外的另一个选择BaseHTTPMiddleware很方便但它会把响应 body 读入内存多一层缓冲和延迟。如果中间件只是要做简单的 header 处理、请求日志、链路追踪可以直接写纯 ASGI 中间件class HeaderInjectionMiddleware: def __init__(self, app): self.app app async def __call__(self, scope, receive, send): if scope[type] ! http: await self.app(scope, receive, send) return scope.setdefault(headers, []) scope[headers].append( (bx-internal-flag, b1) ) await self.app(scope, receive, send)在 FastAPI 里通过app.add_middleware(HeaderInjectionMiddleware)接入。纯 ASGI 中间件没有额外缓冲性能更接近框架底层。我实际开发里通常这样判断只是加个 header、记录耗时用纯 ASGI需要真正读取和修改请求体或响应体才用BaseHTTPMiddleware。这个取舍能让高并发服务的响应延迟少掉不少。5.3 Django 自定义 Manager把业务查询封装成语义方法Django 的objects默认提供filter()、all()但这些查询条件在业务里会重复出现。自定义 Manager 可以把“已发布”“某个作者的已发布”这些业务语义封装成方法代码读起来像英文句子class PublishedQuerySet(models.QuerySet): def published(self): return self.filter(is_publishedTrue) def by_author(self, author_id): return self.filter(author_idauthor_id) class Post(models.Model): title models.CharField(max_length100) is_published models.BooleanField(defaultFalse) author models.ForeignKey(Author, on_deletemodels.CASCADE) objects PublishedQuerySet.as_manager()之后就可以直接写Post.objects.published().by_author(1)比在每个视图里复制filter(is_publishedTrue)要整齐得多。以后业务规则变化比如要顺便排除软删除记录只需要修改这一个方法。这个自定义 Manager 的思路和前面 FastAPI 的依赖工厂其实是同一类模式把规则事先声明好调用方只表达“我要什么”不关心“怎么查”。最后聊一点自己的体会。这些技巧带来的最大变化不是代码行数减少而是“出错的概率”下降了。以前权限校验散落在各个路由里现在收敛到一个依赖工厂以前库存扣减靠临时想起加锁现在用select_for_update成了肌肉记忆以前批量导入数据要跑十几分钟现在一条bulk_create秒完。你不需要一口气把全部技巧都用上不妨挑一个当下最痛的场景先改起来。改完你会体验到原来 Python Web 项目也可以写得既优雅又强悍。 
长河君 · 2026/10/19 6:30:13 · 阅读 1.2k FastAPI与Django隐藏技巧：依赖注入、ORM优化与并发控制实战
做 Web 开发三年我最怕的不是需求变更而是同一个项目写到第 6 个月后代码开始“发臭”路由函数越堆越长ORM 查询到处都是 N1并发一上来库存就超卖。很多人把锅甩给“架构”但我后来发现FastAPI 和 Django 里其实藏着一批不出名但异常好用的 Python 骚操作你不需要重写框架只需要换几种写法就能让项目变得既优雅又强悍。这篇东西不是什么入门教程而是我在几个实际项目里压箱底的隐藏技巧集合适合已经写过 FastAPI/Django 的朋友拿来即用。每一节我都会拆开讲原理、写法和踩坑点保证不是那种复制粘贴就能糊弄过去的八股文。1. 依赖注入不只用来拿参数FastAPI 的 Depends 高级用法很多人用 FastAPI 的Depends只做两件事拿数据库会话、拿当前登录用户。这个用法没错但太浪费了。FastAPI 的依赖注入实际上是一个完整的“依赖图”系统它允许你定义一个“返回依赖的函数”。也就是说Depends的参数不必非得是固定的函数它可以是一个函数工厂请求进来时临时生成一个专属校验器。1.1 带参数的依赖工厂同一套代码管理不同权限假设你要做权限控制最常见的想法是写一个require_admin、一个require_editor、一个require_viewer。三个函数逻辑几乎一样只是权限字符串不同代码重复不说以后加权限还要到处复制。用依赖工厂就可以把“权限字符串”变成参数from typing import Annotated from fastapi import Depends, HTTPException, status def require_permission(permission: str): def checker( user: Annotated[User, Depends(get_current_user)] ) - User: if permission not in user.permissions: raise HTTPException( status_codestatus.HTTP_403_FORBIDDEN, detailInsufficient permissions ) return user return checker app.get(/admin) def admin_dashboard( current_user: Annotated[User, Depends(require_permission(admin))] ): return {msg: fHello {current_user.name}} app.put(/posts/{post_id}) def edit_post( current_user: Annotated[User, Depends(require_permission(editor))] ): ...这里的关键点是require_permission内部返回的checker函数自己还可以声明依赖get_current_user。FastAPI 会解析整条依赖链并且checker里拿到的用户对象就是真正的当前用户。我从这套写法里吃到最大的甜头不是少写几个函数而是权限规则收敛到了一个地方以后想加“超级管理员”权限不用再写新依赖直接require_permission(super_admin)路由签名上也能直观看到权限要求。1.2 子依赖与依赖缓存Depends是可以嵌套的而且 FastAPI 默认会在同一个请求里缓存依赖的解析结果。这意味着哪怕同一个依赖被十个下游同时声明实际也只执行一次。async def get_current_user(token: Annotated[str, Depends(oauth2_scheme)]): user await cache_or_db_get_user(token) return user async def get_user_team(user: Annotated[User, Depends(get_current_user)]): return await get_team_of_user(user.id) app.get(/me/summary) async def get_summary( user: Annotated[User, Depends(get_current_user)], team: Annotated[Team, Depends(get_user_team)], ): ...在这个例子里get_user_team依赖get_current_user路由又显式依赖get_current_user因为 FastAPI 的use_cache参数默认是True所以get_current_user在整个请求生命周期里只会被调用一次。这是一笔很划算的账十个下游依赖各自声明“我要用户”数据库只查一次。如果你确实希望每次调用都重新执行依赖可以设置Depends(..., use_cacheFalse)。不过我用 FastAPI 两年真正需要关闭缓存的情况少之又少大部分时候默认行为反而帮我省掉了大量重复查询。1.3 给依赖套上 lru_cache把只读查询变成内存加速还有一种更极致的缓存玩法。如果依赖函数是完全不依赖请求上下文的纯函数比如读配置项、查固定的字典表可以直接用functools.lru_cache做进程内缓存from functools import lru_cache lru_cache(maxsize128) def get_config_by_key(key: str) - str: return db.query(Config).filter(Config.key key).first().value之后路由里写Depends(get_config_by_key(site_name))这个值在每个进程里最多查一次库。这里有几个必须记住的坑lru_cache的缓存 key 是参数所以参数必须在业务上不可变不要对依赖请求上下文的函数使用lru_cache否则request、session会被长期持有内存迟早爆掉如果需要按时间过期换cachetools.TTLCache或直接查项目里已有的 Redis总之不要硬套这个装饰器。提示这一节所有技巧的共同点是让 FastAPI 替你管理依赖关系和生命周期。依赖负责拿数据和校验路由只负责组织业务你的代码自然就清爽了。2. Pydantic v2 校验器的黑科技把脏活累活拦截在业务代码之外Python Web 项目逃不开数据校验。FastAPI 天生跟 Pydantic 深度绑定Django 里也有各种 validator但很多人只把 Pydantic 用在“声明字段类型”这一层。Pydantic v2 在Annotated和校验器上提供了非常多隐藏能力足够你把数据清洗、格式转换、跨字段校验全部放到接口层业务代码里拿到的都是“已经可信的数据”。2.1 Annotated 类型别名让路由签名瘦成一道闪电先看一段常见的路由签名app.get(/items) def list_items( page: int Query(1, ge1, le1000), size: int Query(20, ge1, le100), ): ...如果多个接口都要分页参数每个路由都要重复写一遍ge1, le1000。用Annotated做类型别名可以把约束声明一次到处复用from typing import Annotated, Literal from fastapi import Query PageNumber Annotated[int, Query(ge1, le1000)] PageSize Annotated[int, Query(ge1, le100)] SortOrder Annotated[Literal[asc, desc], Query()] app.get(/items) def list_items( page: PageNumber 1, size: PageSize 20, order: SortOrder asc, ): ...好处是显而易见的分页规则改一处所有接口同步生效。我特别想强调Annotated不是 FastAPI 特有的语法它是 Python 3.9 加入typing的标准能力FastAPI 只是识别了里面挂载的Query、Path、Depends这些元数据。你完全可以在自己的业务函数里用Annotated做参数标注比如“这个参数必须经过脱敏处理”让类型系统帮你记住约定。2.2 BeforeValidator 与 AfterValidator数据还没定型时先动手Pydantic v2 的BeforeValidator允许你在 Pydantic 做类型转换之前先对原始值做一次处理。这非常适合“容忍脏数据”的接口。举个例子前端可能传12%这种带百分号的税率后端想存小数0.12from typing import Annotated from pydantic import BaseModel, BeforeValidator def parse_percent(value) - float: if isinstance(value, str) and value.endswith(%): return float(value[:-1]) / 100 return float(value) PercentFloat Annotated[float, BeforeValidator(parse_percent)] class TaxRule(BaseModel): rate: PercentFloat请求体传{rate: 5%}Pydantic 会先调parse_percent拿到0.05再去校验类型。如果传了abcfloat(abc)抛ValueErrorFastAPI 自动转成 422。AfterValidator则是在类型转换完之后再执行适合“清洗已转换的数据”比如去掉字符串首尾空格、压缩多个空格。单个看这些功能很碎组合起来就是输入层的堡垒任何格式不干净的数据在进业务代码之前就已经被纠正了。2.3 model_validator跨字段的联动校验field_validator只能看一个字段跨字段逻辑就得用model_validator。最典型的场景是注册时的密码确认from pydantic import BaseModel, model_validator class RegisterForm(BaseModel): username: str password: str confirm_password: str model_validator(modeafter) def passwords_match(self): if self.password ! self.confirm_password: raise ValueError(password and confirm_password do not match) return selfmodeafter表示所有字段校验完成后才执行这段代码所以可以放心读取self.password和self.confirm_password。这种写法比在视图函数里手动if form.password ! form.confirm_password干净得多而且错误能统一走 Pydantic 的校验错误格式返回前端解析逻辑完全一致。2.4 ConfigDict 里的隐藏开关Pydantic v2 里ConfigDict有很多“开关”我项目里最常用到的是这几个配置项效果典型场景str_strip_whitespaceTrue所有字符串字段自动去掉首尾空格表单、用户名extraforbid请求体里出现未声明字段直接报错防止接口参数悄悄变化validate_assignmentTrue实例属性被赋值时重新校验后端修改模型字段时的保险validate_defaultTrue字段默认值也参与校验带默认值的配置类看一个具体用法from pydantic import BaseModel, ConfigDict class CreateUser(BaseModel): model_config ConfigDict( str_strip_whitespaceTrue, extraforbid, ) username: str email: str如果请求体传{username: foo , email: ab.com, role: admin}在extraforbid下会直接报 422而不是静默忽略掉role字段。对接口兼容性来说这个保护非常关键至少在字段悄悄变化时你能第一时间发现。2.5 response_model_exclude_unset响应体别泄露默认值再分享一个响应层面的技巧。假设你的ItemOut模型里discount字段默认是0.0但某个创建接口根本没设置它。如果直接返回整个模型前端就会看到discount: 0.0这会误导调用方。加上response_model_exclude_unsetTrueFastAPI 只会返回显式传过或显式赋值的字段app.post(/items, response_modelItemOut, response_model_exclude_unsetTrue) async def create_item(item: ItemIn): return ItemOut(**item.model_dump(), discount0.0)这个参数在做“部分更新”的 PATCH 接口时特别实用。以前我为了隐藏默认值手写过一堆 dict 过滤代码后来发现 FastAPI 天生就有这个能力直接省掉十几行重复劳动。3. Django ORM 的隐藏加速键Prefetch、F() 与批量更新的组合拳如果说 FastAPI 的优雅来自“依赖与类型系统”那 Django 的强悍就体现在 ORM 的细节里。我见过太多 Django 项目跑慢了不是数据库不行而是代码里全是 N1 查询和逐条 UPDATE。下面这几个技巧全是我优化了多个后台项目后沉淀下来的核心思路只有一句话能一条 SQL 解决的事情绝不用一百条。3.1 Prefetch 对象按条件预取而不是全量预取很多开发者知道prefetch_related可以一次查出关联对象但默认是全量预取。如果每个作者的帖子有几千条预取本身也会变成灾难。Prefetch对象允许你定制预取的 QuerySet 和属性名from django.db.models import Prefetch recent_posts Post.objects.filter( is_publishedTrue ).order_by(-created_at)[:3] authors Author.objects.prefetch_related( Prefetch(posts, querysetrecent_posts, to_attrrecent_posts) ).all() for author in authors: for post in author.recent_posts: print(author.name, post.title)这里to_attrrecent_posts把预取结果挂到对象上名字可以随便取关键是只有最近 3 篇会被加载。注意这个预取结果是 Python list不是 QuerySet不能再继续.filter()。如果后续要过滤要么换字段名要么在内存循环里处理。还有一个坑Prefetch的queryset不能带聚合和复杂的注解在部分数据库后端会报错或静默返回错误数据我建议先用真实的数据库方言跑一遍再上线。3.2 bulk_create 与 bulk_update批量操作的正确姿势导入一万条数据时很多人还是for obj in objs: obj.save()。这会生成一万次 SQL 往返而bulk_create常常一条 SQL 就能完成products [ Product(namefsku-{i}, pricei * 10) for i in range(10_000) ] Product.objects.bulk_create(products, batch_size500)批量更新则略有不同得先把对象加载进内存、改好字段再一次提交products Product.objects.filter(price__lt100) for p in products: p.price 10 Product.objects.bulk_update(products, [price], batch_size500)这里最容易被忽略的坑是bulk_create和bulk_update都不会触发模型的save()方法所以post_save信号、自定义逻辑全部失效。如果你依赖信号去同步搜索索引或缓存批量操作会让你的“钩子”悄悄失灵。我的习惯是在批量操作之后手动执行一次类似refresh_search_index()的清理函数把需要同步的数据显式刷新一遍。3.3 F() 表达式让 UPDATE 在数据库端原子完成计数器加一这个操作最反面的写法是article Article.objects.get(pkpk) article.views 1 article.save()这是典型的竞态条件两个请求同时读到views10各自加一写回最终变成 11 而不是 12。正确姿势是用F()表达式from django.db.models import F Article.objects.filter(pkpk).update(viewsF(views) 1)F()会在数据库端完成SET views views 1读取和更新在同一个 SQL 语句里完成不会因为 Python 层的读取延迟导致少算。类似的扣减库存可以用F(stock) - quantity更新时间戳可以直接用F(last_seen)。再进阶一点还可以配合Case、When做条件式原子更新from django.db.models import Case, When, Value Article.objects.filter(pkpk).update( statusCase( When(views__gte1000, thenValue(hot)), defaultValue(normal), ) )这一招把“先查后改”变成了“一把 UPDATE 搞定”既省往返又避免并发脏读是我做统计面板时最常用的一招。3.4 annotate Window分组排名不再需要子查询“每个分类下热度前三的帖子”这种需求以前要么三次子查询要么把数据全查回 Python 里排序。Django 的Window函数配合RowNumber可以一步搞定from django.db.models import Window, F from django.db.models.functions import RowNumber top_posts Post.objects.annotate( rnWindow( expressionRowNumber(), partition_byF(category), order_byF(views).desc(), ) ).filter(rn__lte3)partition_by相当于给数据分组order_by决定组内排序RowNumber()打出组内的行号外面再套一层filter(rn__lte3)就是每组前三。这里有个前置条件窗口函数需要数据库支持。PostgreSQL 和 MySQL 8.0 以上都没问题SQLite 3.25 之后才支持老版本 SQLite 会直接报错所以测试环境最好用真实目标数据库跑一遍不要只依赖本地 SQLite。4. 事务与并发的优雅博弈锁、提交后回调与异步限流Web 项目做到后面真正让人头疼的往往不是增删改查而是多个请求同时改同一个资源的问题。这一节集中聊聊我在项目里处理并发时的几个固定姿势。4.1 select_for_update库存扣减不再超卖经典的秒杀/下单场景库存只有 10 件100 个人同时抢。如果先查库存再减库存必然超卖。select_for_update可以在数据库层面把选中的行锁住直到事务结束from django.db import transaction def deduct_stock(product_id: int, quantity: int): with transaction.atomic(): product Product.objects.select_for_update().get(pkproduct_id) if product.stock quantity: raise ValueError(库存不足) product.stock - quantity product.save()几个关键点必须记牢。第一select_for_update必须在事务内所以整体要包在transaction.atomic()里。第二锁是有代价的事务里只做必要的查询和更新绝对不要在里面调外部 API 或做耗时的计算否则锁持有时间太长并发能力直线下降。第三如果不想一直等被锁的行可以加nowaitTrue拿不到锁直接抛异常由上层重试。还有一个容易踩的坑如果你在事务里先普通get了一次再select_for_update().get第一次查出来的对象不带锁第二次才真正锁定。正确做法是一开始就写select_for_update().get()不要多此一举。4.2 transaction.on_commit信号别半路跑飞Django 的信号post_save等默认是在事务执行过程中触发的。这带来一个很隐蔽的坑事务里发了post_save紧接着事务回滚监听函数拿到的 instance 其实并没有真正入库。比如注册用户时发欢迎邮件邮件发出去了用户创建事务却回滚了结果用户不存在邮件却发给了不存在的账号。transaction.on_commit就是专门解决这个问题的它把回调推迟到事务真正提交之后from django.db import transaction def user_post_save(sender, instance, created, **kwargs): if created: transaction.on_commit( lambda: send_welcome_email(instance.id) ) post_save.connect(user_post_save, senderUser)这样只有User.objects.create()所在的transaction.atomic()成功提交邮件才会发送。我在 FastAPI 项目里也沿用了这个思路把“发消息、清缓存、更新搜索引擎”这类副作用放到数据库事务提交成功之后再执行。这个模式能避免非常多“数据没存上但副作用已经发生”的诡异问题。4.3 asyncio.gather 与 Semaphore既快又不被打挂FastAPI 是异步框架很多人喜欢用asyncio.gather并发请求外部 API。如果没有控制并发量1000 个 URL 同时打出去下游服务很可能直接 503。Semaphore可以给并发套上一个“阀门”import asyncio import httpx async def fetch_one(client: httpx.AsyncClient, url: str, limiter: asyncio.Semaphore): async with limiter: resp await client.get(url) resp.raise_for_status() return resp.json() async def fetch_many(urls: list[str]): limiter asyncio.Semaphore(10) async with httpx.AsyncClient(timeout10) as client: tasks [fetch_one(client, url, limiter) for url in urls] results await asyncio.gather(*tasks, return_exceptionsTrue) return results这里有几个细节。limiter在fetch_many内部创建避免了 Python 3.10 以下版本跨事件循环使用Semaphore的兼容问题。return_exceptionsTrue也很重要某个请求失败不会中断其他请求你可以拿到异常结果后统一做重试或记录。关于并发上限没有万能答案我一般先压测下游接口找到它稳定响应的 TPS然后把Semaphore初始值设为 TPS 的一半观察延迟曲线再调整。先小后大、逐步调优才是稳妥路径。5. 项目组织的魔法生命周期、中间件与自定义 Manager前面讲的都是“单点技巧”但一个项目要长期可维护还需要结构层的“魔法”。这个部分聊的是代码组织怎么把全局资源、公共逻辑、业务查询收拢到合适的位置。5.1 lifespan用异步上下文管理器的思路管理 FastAPI 生命周期很多老项目还在用app.on_event(startup)初始化全局资源这个写法在新版 FastAPI 里已经不推荐了。更好的方式是lifespan它把启动和关闭逻辑收拢到一个上下文管理器里from contextlib import asynccontextmanager from fastapi import FastAPI asynccontextmanager async def lifespan(app: FastAPI): app.state.pool await create_connection_pool() app.state.cache await create_cache_client() yield await app.state.pool.close() await app.state.cache.close() app FastAPI(lifespanlifespan)启动时创建连接池、预热缓存结束时优雅回收一目了然。这比on_event强在你想让哪个资源什么时候创建、什么时候释放全都写在同一个函数里逻辑不会散落在多个装饰器之间。尤其配合app.state存全局对象路由里用request.app.state.pool取用依赖关系非常清晰。5.2 中间件BaseHTTPMiddleware 之外的另一个选择BaseHTTPMiddleware很方便但它会把响应 body 读入内存多一层缓冲和延迟。如果中间件只是要做简单的 header 处理、请求日志、链路追踪可以直接写纯 ASGI 中间件class HeaderInjectionMiddleware: def __init__(self, app): self.app app async def __call__(self, scope, receive, send): if scope[type] ! http: await self.app(scope, receive, send) return scope.setdefault(headers, []) scope[headers].append( (bx-internal-flag, b1) ) await self.app(scope, receive, send)在 FastAPI 里通过app.add_middleware(HeaderInjectionMiddleware)接入。纯 ASGI 中间件没有额外缓冲性能更接近框架底层。我实际开发里通常这样判断只是加个 header、记录耗时用纯 ASGI需要真正读取和修改请求体或响应体才用BaseHTTPMiddleware。这个取舍能让高并发服务的响应延迟少掉不少。5.3 Django 自定义 Manager把业务查询封装成语义方法Django 的objects默认提供filter()、all()但这些查询条件在业务里会重复出现。自定义 Manager 可以把“已发布”“某个作者的已发布”这些业务语义封装成方法代码读起来像英文句子class PublishedQuerySet(models.QuerySet): def published(self): return self.filter(is_publishedTrue) def by_author(self, author_id): return self.filter(author_idauthor_id) class Post(models.Model): title models.CharField(max_length100) is_published models.BooleanField(defaultFalse) author models.ForeignKey(Author, on_deletemodels.CASCADE) objects PublishedQuerySet.as_manager()之后就可以直接写Post.objects.published().by_author(1)比在每个视图里复制filter(is_publishedTrue)要整齐得多。以后业务规则变化比如要顺便排除软删除记录只需要修改这一个方法。这个自定义 Manager 的思路和前面 FastAPI 的依赖工厂其实是同一类模式把规则事先声明好调用方只表达“我要什么”不关心“怎么查”。最后聊一点自己的体会。这些技巧带来的最大变化不是代码行数减少而是“出错的概率”下降了。以前权限校验散落在各个路由里现在收敛到一个依赖工厂以前库存扣减靠临时想起加锁现在用select_for_update成了肌肉记忆以前批量导入数据要跑十几分钟现在一条bulk_create秒完。你不需要一口气把全部技巧都用上不妨挑一个当下最痛的场景先改起来。改完你会体验到原来 Python Web 项目也可以写得既优雅又强悍。