开场你遇到过这种事前端同学催接口你用 Flask 写了个 POST 接口curl 测了一下返回 500。查日志、查文档、改代码、再测还是 500。问了组里没人有空帮你 Debug你自己在那折腾了一下午。这种场景下 FastAPI 才是正确答案。它不只是一个更快的 Flask它是一套完整的请求-验证-响应闭环。写完到调通慢的话十分钟快的话三分钟。这篇文章假设你会 Python 语法写过 Django 或 Flask或者跟着教程做过小 demo想知道怎么用 FastAPI 搭一个能上线的接口。我不会给你一个Hello World然后截个 Swagger UI 截图就结束而是带你走一遍从写接口到部署的全流程中间哪些地方容易踩坑、哪些地方有判断都说清楚。一、为什么是 FastAPI先说清楚 FastAPI 解决了什么问题再说为什么不是 Django 或 Flask。自动生成 OpenAPI 文档。你写完接口FastAPI 自动生成一份 Swagger UI跑在/docs路径下。前端工程师可以直接在这个页面上点Try it out实时调试接口不用等你写接口文档。这是一个工程上省大量沟通成本的功能不是噱头。Pydantic 数据验证。你定义一个 Pydantic 模型FastAPI 自动验证请求体类型错误直接返回 422 Unprocessable Entity不会让你的接口抛出 500。Django 和 Flask 收到非法数据要么你手动写验证逻辑要么直接炸。手动验证是个坑新手要么不写要么写一半漏一半。异步支持。FastAPI 基于 Starlette天然支持async def。在 I/O 密集场景数据库查询、外部 API 调用下QPS 比 Flask 高出一截。Django 在 3.1 之后也支持 async但生态没有 FastAPI 顺手。Flask 到今天都没有原生异步写长连接只能用 SocketIO 之类的 workaround。和 Starlette 的关系就一句FastAPI 是 Starlette 的超集Starlette 负责底层 HTTP 能力FastAPI 在上面叠了 Pydantic 验证和 OpenAPI 生成。你不需要直接用 Starlette除非你在写一个不想依赖 FastAPI 的轻量库。一个明确观点如果你只需要提供 API 接口不需要模板、不需要 admin 后台、不需要 CMSDjango 的重量完全没必要。FastAPI 只做 API 这一件事反而干净。这里有个实际对比搭一个用户注册 登录 JWT 接口用 Django 需要了解 URL 配置、View 函数、Model 表单、CBV vs FBV、DRF 序列化器前后要写至少四个文件才能跑起来。用 FastAPI一个文件二十行代码类型注解打完Pydantic 模型定义完路由写完依赖注入写完接口就能跑。这个差距不是性能差距是认知负载的差距——你要记的东西越少出错的概率就越低。二、最小可用的接口先跑通再深入。fromfastapiimportFastAPI appFastAPI()app.get(/)defread_root():return{Hello:World}这个文件叫main.py接下来怎么跑。uvicorn main:app--reload命令里main:app的意思是从main.py模块导入名为app的 FastAPI 实例。--reload参数开启热重载你改代码保存进程自动重启不用手动停再起。这个在本地开发阶段非常实用。服务跑起来之后浏览器打开http://127.0.0.1:8000/docs你会看到一个 Swagger UI 页面左侧列表里已经有了GET /这一个接口点开可以直接发请求看返回。这个页面是 FastAPI 自动生成的不需要你写一行代码。踩过的坑直接python main.py会报错。FastAPI 应用不是普通 Python 脚本不能用python命令直接跑必须通过 ASGI 服务器启动。生产环境用 uvicorn 或者 gunicorn 开发环境用uvicorn --reload这是标准做法不是一个可选项。三、请求体Pydantic 怎么用接口要接收 JSON光靠 Flask 那种request.get_json()是不够的你得自己判断字段存不存在、类型对不对。Pydantic 把这件事自动化了。fromfastapiimportFastAPIfrompydanticimportBaseModel appFastAPI()classItemCreate(BaseModel):name:strprice:floatis_available:boolTruetag:str|NoneNoneBaseModel的子类就是你的数据模型。字段名右边是类型注解 True或 None是默认值不写默认值就是必填。现在写一个接收这个模型的接口fromfastapiimportFastAPIfrompydanticimportBaseModelfromenumimportEnum appFastAPI()classTagEnum(str,Enum):electronicselectronicsclothingclothingfoodfoodclassItemCreate(BaseModel):name:strprice:floatis_available:boolTruetag:TagEnum|NoneNoneapp.post(/items/)defcreate_item(item:ItemCreate):# item 参数的类型注解是关键FastAPI 自动解析请求体并验证return{name:item.name,price:item.price,tag:item.tag}当你发一个请求{name: Laptop, price: 999.99, tag: electronics}FastAPI 解析并验证这个 JSON类型全对才进到函数里。如果发{name: Laptop, price: not a number}返回 422{detail:[{loc:[body,price],msg:Input should be a valid number, unable to parse string as a number,type:float_type}]}422 会告诉你具体哪个字段、什么错误。这个信息对前端调试接口非常有用。踩过的坑字段类型写错返回 422 而不是 500傻傻 Debug。有些新手看到返回 422以为是服务器崩了其实恰恰相反——FastAPI 正确地拦截了非法数据并给出了明确的错误信息。422 是正常响应说明你的接口在正常工作。四、数据库SQLAlchemy 2.0 async现在要把数据存到数据库里。选 SQLAlchemy 2.0 async因为这是当前的主流组合能发挥 FastAPI 异步的最大优势。为什么用 asyncFlask 的数据库操作是同步的发起一次查询当前线程会阻塞等待数据库返回。在 QPS 低的场景下这不是问题但当你需要同时处理大量并发请求同步数据库会成为瓶颈。async 操作在等待 I/O 的同时可以让出线程去处理其他请求吞吐量自然就上去了。定义 ModelfromsqlalchemyimportString,Boolean,Floatfromsqlalchemy.ormimportDeclarativeBase,Mapped,mapped_columnclassBase(DeclarativeBase):passclassItem(Base):__tablename__itemsid:Mapped[int]mapped_column(primary_keyTrue)name:Mapped[str]mapped_column(String(255))price:Mapped[float]mapped_column(Float)is_available:Mapped[bool]mapped_column(Boolean,defaultTrue)tag:Mapped[str|None]mapped_column(String(100),nullableTrue)Mapped[type] mapped_column(...)是 SQLAlchemy 2.0 的新写法比旧版的Column(String)更直观类型信息和列定义绑在一起。String(255)是字符上限nullableTrue对应数据库里的 NOT NULL 约束。数据库连接配置fromsqlalchemy.ext.asyncioimportcreate_async_engine,AsyncSession,async_sessionmaker DATABASE_URLsqliteaiosqlite:///./items.dbenginecreate_async_engine(DATABASE_URL,echoTrue)async_session_makerasync_sessionmaker(engine,class_AsyncSession,expire_on_commitFalse)asyncdefget_db():asyncwithasync_session_maker()assession:yieldsessioncreate_async_engine创建异步引擎URL 前缀sqliteaiosqlite://对应 SQLite 的异步驱动。get_db是一个生成器函数FastAPI 的依赖注入系统会自动调用它给每个请求分配一个独立的数据库 session请求结束后自动关闭。这是 FastAPI 最常用的数据库 session 管理方式。CRUD 示例fromfastapiimportFastAPI,Depends,HTTPExceptionfromsqlalchemyimportselectfromsqlalchemy.ext.asyncioimportAsyncSession appFastAPI()# 增asyncdefcreate_item(db:AsyncSession,item:ItemCreate):db_itemItem(nameitem.name,priceitem.price,is_availableitem.is_available,tagitem.tag)db.add(db_item)awaitdb.commit()awaitdb.refresh(db_item)returndb_item# 查所有asyncdefget_items(db:AsyncSession,skip:int0,limit:int100):resultawaitdb.execute(select(Item).offset(skip).limit(limit))returnresult.scalars().all()# 按 ID 查asyncdefget_item(db:AsyncSession,item_id:int):resultawaitdb.execute(select(Item).where(Item.iditem_id))itemresult.scalar_one_or_none()ifnotitem:raiseHTTPException(status_code404,detailItem not found)returnitem# 改asyncdefupdate_item(db:AsyncSession,item_id:int,item_update:ItemCreate):resultawaitdb.execute(select(Item).where(Item.iditem_id))db_itemresult.scalar_one_or_none()ifnotdb_item:raiseHTTPException(status_code404,detailItem not found)forkey,valueinitem_update.model_dump(exclude_unsetTrue).items():setattr(db_item,key,value)awaitdb.commit()awaitdb.refresh(db_item)returndb_item# 删asyncdefdelete_item(db:AsyncSession,item_id:int):resultawaitdb.execute(select(Item).where(Item.iditem_id))db_itemresult.scalar_one_or_none()ifnotdb_item:raiseHTTPException(status_code404,detailItem not found)awaitdb.delete(db_item)awaitdb.commit()return{message:deleted}注意这里用了item_update.model_dump(exclude_unsetTrue)。exclude_unsetTrue的作用是只取前端实际传入的字段没传的字段保持数据库里的原值不被覆盖。这比手动判断每个字段是否为 None 再决定要不要更新要干净得多。关于事务一次commit失败不会影响之前已经commit的操作。如果你的业务逻辑需要多个写操作在同一个事务里完成应该用async with session.begin():把它们包起来这样任何一个操作失败都会回滚所有操作。SQLAlchemy 2.0 的事务 API 比旧版清晰很多习惯了这个写法之后不容易漏掉 rollback。踩过的坑同步 session 和异步 session 混用报 AttributeError。你的 engine 是create_async_enginesession 就得用AsyncSession不能用Session。反过来也一样。混用的话运行时会报错信息大概是AttributeError: coroutine object has no attribute xxx意思是某个方法被当成了协程而不是同步方法。这种错误报得诡异但只要检查create_async_engine和 session 类型是否匹配就能定位。五、路由拆分项目怎么组织一个人写的时候全塞在一个main.py里没问题三个人协作的时候你试试改一个字段定义要等三个人都提交完才能跑通。一个合理的目录结构app/ ├── __init__.py ├── main.py ├── database.py ├── models.py ├── schemas.py ├── crud.py └── routers/ ├── __init__.py └── items.pyapp/main.py是应用入口fromfastapiimportFastAPIfromapp.routersimportitems appFastAPI()app.include_router(items.router)app/routers/items.py里写具体的路由fromfastapiimportAPIRouter,Depends,HTTPExceptionfromsqlalchemy.ext.asyncioimportAsyncSessionfromapp.databaseimportget_dbfromapp.schemasimportItemCreate,ItemResponsefromapp.crudimportcreate_item,get_item,get_items,update_item,delete_item routerAPIRouter(prefix/items,tags[items])router.post(/,response_modelItemResponse,status_code201)asyncdefcreate(item:ItemCreate,db:AsyncSessionDepends(get_db)):returnawaitcreate_item(db,item)router.get(/,response_modellist[ItemResponse])asyncdeflist_items(skip:int0,limit:int100,db:AsyncSessionDepends(get_db)):returnawaitget_items(db,skip,limit)router.get(/{item_id},response_modelItemResponse)asyncdefget_one(item_id:int,db:AsyncSessionDepends(get_db)):returnawaitget_item(db,item_id)router.put(/{item_id},response_modelItemResponse)asyncdefupdate(item_id:int,item:ItemCreate,db:AsyncSessionDepends(get_db)):returnawaitupdate_item(db,item_id,item)router.delete(/{item_id})asyncdefremove(item_id:int,db:AsyncSessionDepends(get_db)):returnawaitdelete_item(db,item_id)APIRouter把一组相关路由打包prefix参数自动给所有路由加上路径前缀tags参数在 Swagger UI 里分组显示。response_model指定返回数据用哪个 Pydantic 模型做序列化自动过滤掉模型里没定义的字段——你不需要手动处理返回格式。app/schemas.py里定义 Pydantic 模型app/crud.py里写数据库操作函数路由层只负责接收请求、调用 CRUD、返回结果三层职责清晰。踩过的坑循环导入。假设app.main导入了app.routers.items而app.routers.items又导入了app.modelsapp.models又导入了app.databaseapp.database导入了app.modelsPython 在解析这些模块的时候还没初始化完任何一个就会报ImportError。解决办法把导入尽量往后放或者把共享的类型定义单独放到一个app/models/base.py里让database.py只导入这个 base阻断循环路径。循环导入在多人协作的项目里是高发问题一旦出现就用这个套路排查。六、分页、过滤、排序列表接口是后端最常见的场景之一数据量大了不可能一次返回所有记录。fromsqlalchemyimportselect,asc,descasyncdefget_items(db:AsyncSession,skip:int0,limit:int100,tag:str|NoneNone,min_price:float|NoneNone,max_price:float|NoneNone,sort_by:strid,order:strasc):queryselect(Item)iftag:queryquery.where(Item.tagtag)ifmin_priceisnotNone:queryquery.where(Item.pricemin_price)ifmax_priceisnotNone:queryquery.where(Item.pricemax_price)sort_columngetattr(Item,sort_by,Item.id)iforderdesc:queryquery.order_by(desc(sort_column))else:queryquery.order_by(asc(sort_column))queryquery.offset(skip).limit(limit)resultawaitdb.execute(query)returnresult.scalars().all()skip/limit是分页的标配offsetlimit对应 SQL 里的LIMIT ? OFFSET ?。过滤条件用where拼接tag用精确匹配价格区间用和。排序用order_by方向由asc和desc控制。getattr(Item, sort_by, Item.id)这个写法允许调用方传字段名字符串但当字段不存在时默认按 id 排序避免 SQL 注入和字段名拼写错误导致查询失败。路由层接这些参数router.get(/,response_modellist[ItemResponse])asyncdeflist_items(skip:int0,limit:int100,tag:str|NoneNone,min_price:float|NoneNone,max_price:float|NoneNone,sort_by:strid,order:strasc,db:AsyncSessionDepends(get_db)):returnawaitget_items(db,skip,limit,tag,min_price,max_price,sort_by,order)调用方可以这样请求GET /items/?tagelectronicsmin_price100max_price5000sort_bypriceorderdesclimit20。踩过的坑filter 里字段写错不会报错查出来是空数组。SQLAlchemy 在构造查询时如果你在where里用了模型里不存在的字段名Python 不会报错它会把这个条件翻译成 SQL 里一个永远为假的表达式类似WHERE 10查出来的结果集是空的但你不会得到任何错误信息。只有在数据库层面你才能看到这个字段不存在Python 层完全沉默。所以 filter 里字段名写错了最典型的表现就是接口返回空数组而不是报错。写 filter 条件时建议对着模型定义核对一遍字段名。七、错误处理优雅地返回 4xx接口出错不可怕乱返回码才可怕。404 找不到返回 200、前端参数错了返回 500这些都是给自己挖坑。FastAPI 提供HTTPException处理常规 HTTP 错误fromfastapiimportHTTPExceptionrouter.get(/{item_id},response_modelItemResponse)asyncdefget_one(item_id:int,db:AsyncSessionDepends(get_db)):resultawaitdb.execute(select(Item).where(Item.iditem_id))itemresult.scalar_one_or_none()ifnotitem:raiseHTTPException(status_code404,detailfItem{item_id}not found)returnitem这个函数在 item 不存在时抛出一个 HTTPExceptionFastAPI 会把它转换成正确的 HTTP 响应{detail:Item 42 not found}常见的错误码用法404 Not Found资源不存在400 Bad Request请求参数本身不合法比如 min_price 大于 max_price403 Forbidden用户无权访问这个资源422 Unprocessable Entity请求体验证失败FastAPI 自动处理 Pydantic 验证错误不需要你手动抛有时候你需要统一处理某些错误类型比如所有SQLAlchemyNoResultFound都返回 404。可以注册一个全局异常处理器fromfastapiimportRequestfromfastapi.responsesimportJSONResponseapp.exception_handler(HTTPException)asyncdefhttp_exception_handler(request:Request,exc:HTTPException):returnJSONResponse(status_codeexc.status_code,content{detail:exc.detail,path:str(request.url)})这个处理器给所有 HTTPException 响应加了一个path字段方便前端排查是哪个接口出了问题。踩过的坑直接raise Exception会变成 500。在 FastAPI 的路由函数里如果你写raise Exception(something went wrong)FastAPI 会捕获这个未处理的异常返回 500 Internal Server Error而不是你想要的 403 或 400。原因是 FastAPI 只识别HTTPException和你在app.exception_handler里注册过的异常类型其他异常一律按 500 处理。所以如果需要返回特定状态码一定要用HTTPException不要用普通异常。八、部署uvicorn systemd开发阶段用uvicorn main:app --reload够了生产环境要严肃对待。uvicorn 启动参数uvicorn app.main:app--host0.0.0.0--port8000--workers4--host 0.0.0.0监听所有网卡这样外网才能访问。--port 8000指定端口。--workers 4启动 4 个 worker 进程每个 worker 独立运行一个 uvicorn 实例可以利用多核 CPU 并行处理请求。什么时候用 gunicorn uvicorn workersuvicorn 自身不支持多 worker 时做进程管理。如果你的应用是纯异步的所有 I/O 操作都是 async可以直接用uvicorn --workers N。如果你的应用混了同步代码或者需要更好的进程管理和信号处理用 gunicorn 作为进程管理器uvicorn 作为 worker 类型gunicorn app.main:app-w4-kuvicorn.workers.UvicornWorker-b0.0.0.0:8000gunicorn 会管理这 4 个 worker 进程的启动、重启、优雅退出比直接用 uvicorn workers 稳定性更好。systemd 守护进程写一个 service 文件让进程挂了能自动重启[Unit] DescriptionFastAPI Items Service Afternetwork.target [Service] Userwww-data Groupwww-data WorkingDirectory/opt/app ExecStart/opt/app/venv/bin/uvicorn app.main:app --host 0.0.0.0 --port 8000 --workers 4 Restartalways RestartSec5 [Install] WantedBymulti-user.target把这段保存为/etc/systemd/system/items.service然后sudosystemctl daemon-reloadsudosystemctlenableitemssudosystemctl start itemssudosystemctl status itemsRestartalways确保进程崩溃后 5 秒自动重启。sudo systemctl status items会显示当前进程状态如果看到active (running)说明一切正常。生产环境还需要处理日志。uvicorn 默认输出到 stdoutsystemd 用journalctl -u items -f可以实时查看日志但这不是持久化日志的方案。如果你的服务器有 logrotate建议在 service 文件里加一行StandardOutputjournal然后配置 journald 的存储策略或者直接重定向到文件ExecStart/opt/app/venv/bin/uvicorn app.main:app --host 0.0.0.0 --port 8000 --workers 4 /var/log/items.log 21重启 systemd 服务后日志就写入文件了配合tail -f /var/log/items.log实时看grep ERROR /var/log/items.log快速定位问题。这两件事看起来是运维细节但线上出了问题你不知道去哪看日志会浪费大量时间。踩过的坑workers 数量设成 CPU 核数的 2 倍内存直接爆了。网上有些文章说workers 2 * CPU核数 1这是针对同步阻塞型 worker 的经验公式用在 uvicorn 的 async worker 上不适用。async worker 在等待 I/O 时不占线程CPU 利用率比同步 worker 低得多设太多 worker 反而会导致内存占用翻倍响应时间变长。实测下来CPU 密集型场景 workers 设为 CPU 核数I/O 密集型场景设为核数的 2 到 3 倍就差不多了具体要结合内存容量和压测结果调整。结尾FastAPI 的核心优势就三个快——开发体验好Swagger UI 自带写接口不用另起文档稳——Pydantic 替你做了数据验证422 比 500 好处理得多强——异步支持让 I/O 密集型接口的 QPS 比 Flask 高出一个量级。搭完接口记得去/docs看看确认接口列表、分组、参数说明都在Try it out 能跑通。这是 FastAPI 给你省下来的工作量花两分钟检查一下值得。