这段时间FastAPI在Python后端圈子里的热度一直很高,我自己也在几个项目里陆续从Flask和Django迁到了FastAPI,整体体验相当不错。今天不聊泛泛的框架介绍,就从一个能落地的FastAPI项目出发,把选型理由、项目结构、CORS配置、SQLAlchemy集成、纯REST接口实现到常见坑位,完整走一遍。无论你是刚接触FastAPI的新手,还是准备把老项目迁过来的同学,这篇文章都值得认真过一遍。
FastAPI解决的核心问题其实很直接:用Python快速写出高性能的REST接口,同时利用类型注解把参数校验、数据序列化、接口文档自动生成这些脏活全干了。适合用来做前后端分离项目的API层、微服务网关、内部工具平台,以及任何对开发效率和接口规范性有要求的场景。接下来说的全部内容,都是我实际项目中验证过的方案,不是网上抄来的demo。
1. 为什么是FastAPI而不是Flask/Django
1.1 性能到底快在哪
很多人听到FastAPI第一反应是"性能好",但具体好在哪里,得拆开看。FastAPI底层是Starlette,Starlette底层又是uvicorn这个ASGI服务器。ASGI和传统WSGI的区别在于它是异步的,uvicorn还额外集成了uvloop和httptools,这两件套能显著提升事件循环和HTTP解析的效率。
打个简单比方:传统的Flask开发模式,每个请求进来就像是餐厅里一个服务员全程跟着一桌客人,客人多了就得不断加服务员(线程/进程),服务员一多互相抢道,开销就上来了。而FastAPI的异步模型更像一个高效的前台调度员,一个人同时对接多桌客人的点单需求,遇到需要等待的环节(比如查数据库),就先接下一单,等结果回来了再回头处理,人数不用加太多,餐厅吞吐量照样高。
但要注意,这种性能优势主要体现在IO密集场景,比如大量数据库查询、外部API调用、读写缓存。如果你的接口是CPU密集型,比如图像处理、复杂计算,那async帮不了太多,该用Celery队列还是得用。
1.2 类型注解带来的开发体验飞跃
这是我个人最看重的一点。FastAPI把所有东西都建立在Python类型注解之上,请求参数、请求体、响应模型,全部用类型声明。Pydantic在后台完成数据校验和序列化,校验失败自动返回422错误,字段缺失、类型不对、范围超限,都给你写得明明白白。
举个例子,一个创建图书的接口,前端传了{"title": "Python实战", "price": "abc"},Pydantic会直接告诉你price字段期望float,拿到的是str,根本不需要自己在代码里写一堆if判断。省下的这部分工作量,在接口多的时候是非常可观的。
更妙的是,类型声明同时驱动了多个环节:校验逻辑是它,接口文档是它,IDE自动补全也是它。改一个字段类型,IDE里所有用到的地方立刻标红,这种重构信心在Flask那种自由度极高的项目里是感受不到的。
1.3 自动化交互式文档
FastAPI会自动生成OpenAPI规范,然后基于这个规范给你两个现成的文档页面:/docs用的是Swagger UI,/redoc用的是ReDoc。打开http://127.0.0.1:8000/docs,你能看到所有接口的请求参数、请求体示例、响应结构,还能直接在页面上点"Try it out"发起真实请求。
这个能力在后端交付场景里特别实用。以前用Flask的时候,每次联调都靠Postman导来导去,或者手写Markdown文档,改一个字段到处同步。FastAPI的文档是代码自带的,代码改了文档立刻变,永远不会出现"代码和文档不一致"这种经典扯皮问题。
1.4 什么时候不建议用FastAPI
FastAPI不是万能的,踩过坑之后我也有清醒的认知。如果项目是一个高度依赖服务端模板渲染的站点,或者需要一个自带admin后台、ORM、迁移工具全家桶的管理系统,Django依然是更稳妥的选择。如果团队里大部分人都不熟悉异步编程,让他们强制执行async风格,反而会写出比同步更慢的代码。选型要结合团队实际情况,框架再优秀,团队成员学不会、用不好,也是白搭。
2. 环境准备与项目目录结构
2.1 Python版本与安装
FastAPI要求Python 3.8以上,但我个人建议直接用3.11或3.12,从3.11开始Python引入了大量性能优化,跑同样的代码整体会快一截。装好Python之后,建议每个项目都建独立虚拟环境,不要图省事装到全局。
python -m venv venv source venv/bin/activate # Windows下执行 venv\Scripts\activate pip install fastapi "uvicorn[standard]"这里有个细节:uvicorn[standard]一定要带[standard]这部分。standard模式会额外安装uvloop和httptools,性能提升非常明显,实测同样的接口,QPS能差出一个量级。生产环境部署的时候,这个细节很容易被忽略,很多人装成纯uvicorn就上了,性能白丢一大截。
2.2 可维护的项目目录结构
热词里有人专门搜"fastapi项目目录结构",说明这个问题确实困扰了不少人。FastAPI没有强制规定目录长什么样,但项目一大,结构混乱带来的维护成本是实打实的。下面是我个人经过几个项目迭代后沉淀下来的结构,仅供参考,重点是理解每一层为什么存在。
my_fastapi_project/ ├── app/ │ ├── __init__.py │ ├── main.py # 应用入口,创建FastAPI实例,注册路由 │ ├── core/ │ │ ├── config.py # 配置项,读取.env、环境变量 │ │ └── database.py # 数据库引擎和会话管理 │ ├── models/ # SQLAlchemy ORM模型 │ │ └── book.py │ ├── schemas/ # Pydantic模型(请求/响应) │ │ └── book.py │ ├── crud/ # 数据库操作层,复用性高 │ │ └── book.py │ ├── routers/ # API路由层 │ │ └── book.py │ └── dependencies.py # 公共依赖,如get_db ├── .env ├── requirements.txt └── README.md这里我想多说一句设计思路。models、schemas、crud、routers这四层是分离的,每层职责单一:模型层只管数据库表映射,Schema层只管API的输入输出定义,CRUD层只管数据操作逻辑,路由层只管接口定义和参数绑定。这样做的好处是,当你需要新增一张表、一个接口或者改一个校验规则时,改动范围可以被精确控制,不会牵一发动全身。
如果你的项目更大,建议进一步按业务域拆分子包,而不是把所有模型堆在一个models目录里。比如用户域、订单域、商品域各自建包,每个包内包含自己的models、schemas、routers。这是我的核心经验:目录结构跟着业务边界走,而不是跟着技术组件走。
2.3 最小可运行应用
先把最小骨架跑起来,后面再一步步填充细节。
from fastapi import FastAPI app = FastAPI(title="我的第一个FastAPI应用", version="0.1.0") @app.get("/") def read_root(): return {"message": "Hello FastAPI"}保存为main.py,启动:
uvicorn main:app --reload --port 8000--reload参数开启热重载,改代码后服务自动重启,开发期必备。但要注意,热重载只建议在开发环境用,生产环境开热重载会白白消耗资源,还可能因为代码变更触发意外的重启。
启动后访问http://127.0.0.1:8000/docs,你就能看到自动生成的Swagger文档了。到这里别急着高兴,接下来要解决的是实际开发中躲不开的几个核心配置。
3. 跨域配置:CORS的原理与FastAPI实践
3.1 前后端分离为什么会撞上跨域
"fastapi cors"能成为热搜词,说明跨域问题卡住了不少人。简单说,浏览器的同源策略规定,一个页面只能请求同协议、同域名、同端口下的资源。前端开发服务器跑在localhost:3000,后端FastAPI跑在localhost:8000,端口不一样,天然就是跨域。
浏览器遇到跨域请求时会先分情况处理:简单的GET/POST请求直接发出,但带自定义Header、非简单Content-Type的请求,会先发一个OPTIONS预检请求,问服务器"允不允许我这么干",服务器点头后浏览器才发真正的请求。预检请求处理不好,前端就会看到"CORS policy: No 'Access-Control-Allow-Origin' header"这类报错。
3.2 FastAPI中配置CORSMiddleware
FastAPI提供了现成的中间件来处理跨域。你不需要自己写任何有关Header的逻辑,用CORSMiddleware配一下就行。
from fastapi.middleware.cors import CORSMiddleware app.add_middleware( CORSMiddleware, allow_origins=[ "http://localhost:3000", "http://127.0.0.1:3000", ], allow_credentials=True, allow_methods=["GET", "POST", "PUT", "DELETE", "OPTIONS"], allow_headers=["*"], )参数看着简单,但有几个地方非常容易踩坑。
先说allow_origins。开发环境图省事可能直接填["*"],表示允许所有来源。但如果你同时设置了allow_credentials=True,那么浏览器会拒绝这次跨域请求,因为带Cookie的跨域请求不允许Access-Control-Allow-Origin为星号。所以,只要你将来有可能用到Cookie或Session认证,allow_origins就必须写成具体的域名列表。
3.3 实际项目中的CORS注意事项
第一个注意点是allow_credentials=True什么时候用。很多项目用JWT,把token放在Authorization Header里,这种情况下其实allow_credentials可以设False,因为JWT不走Cookie。但如果你做的是传统登录态,用Cookie存Session ID,那allow_credentials必须为True,否则浏览器拿不到Set-Cookie响应头,登录态根本建立不起来。
第二个注意点是,本地开发时localhost和127.0.0.1虽然指向同一台机器,但浏览器把它俩视为不同源。所以配置里两个地址最好都写上,不然前端用localhost访问,你在allow_origins里只写了127.0.0.1,照样报跨域错。
第三个注意点是,中间件的添加顺序。理论上app.add_middleware哪一行加都行,但如果你同时用了多个自定义中间件,要理解中间件是洋葱模型,请求按添加顺序从外到内进入,响应从内到外返回。CORS中间件通常要放在最外层,确保预检请求能第一时间被处理掉。
4. 接入SQLAlchemy构建高性能Web服务
4.1 SQLAlchemy 2.0风格的模型定义
"fastapi和sqlalchemy构建高性能web服务"这个组合是FastAPI项目的主流方案。FastAPI本身不做ORM,数据库操作全靠SQLAlchemy。我用的是SQLAlchemy 2.0风格,和旧版写法有区别,如果你看网上的教程很多还是1.x老风格,注意分辨。
2.0风格用DeclarativeBase来定义基类,模型的列类型用Mapped和mapped_column来声明。
from datetime import datetime from sqlalchemy import String, Float, DateTime, func from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column class Base(DeclarativeBase): pass class Book(Base): __tablename__ = "books" id: Mapped[int] = mapped_column(primary_key=True, index=True) title: Mapped[str] = mapped_column(String(200), index=True) author: Mapped[str] = mapped_column(String(100)) price: Mapped[float] = mapped_column(Float) stock: Mapped[int] = mapped_column(default=0) created_at: Mapped[datetime] = mapped_column(DateTime, server_default=func.now())2.0风格最大的好处是类型安全:Mapped[str]告诉SQLAlchemy这个字段的类型是str,IDE和mypy都能基于这个做静态检查。mapped_column替代了旧的Column,写法上更简洁。server_default=func.now()的意思是让数据库在插入时生成当前时间,这个比Python侧default=datetime.now更可靠,因为数据库层的默认值在直接执行SQL时也生效。
4.2 数据库连接与会话管理
接下来是建立数据库连接。我以MySQL为例,因为这是生产环境最常见的配置。别忘了装对应的驱动,PyMySQL是纯Python的,简单但慢;asyncmy是异步驱动,性能好,和FastAPI的异步模型更搭。
from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker, Session from typing import Generator DATABASE_URL = "mysql+pymysql://root:password@localhost/fastapi_demo?charset=utf8mb4" engine = create_engine( DATABASE_URL, pool_size=10, max_overflow=20, pool_pre_ping=True, ) SessionLocal = sessionmaker(bind=engine, autoflush=False, autocommit=False) def get_db() -> Generator[Session, None, None]: db = SessionLocal() try: yield db finally: db.close()这里有一个非常关键的设计:get_db这个依赖函数。FastAPI的依赖注入系统会在请求进来时调用get_db,当一个请求结束时,无论接口逻辑是否抛异常,finally里的db.close()都会执行,确保数据库连接被释放回连接池。
我见过很多新手项目不搞依赖注入,每个接口里自己SessionLocal()开一个会话,用完忘了close(),跑一段时间后数据库连接池就满了,应用报"Too many connections"错误,排查起来特别痛苦。所以get_db这个模式不是锦上添花,而是必须养成的习惯。
pool_size=10表示连接池里维持10个连接,max_overflow=20表示连接不够用的时候最多额外创建20个。pool_pre_ping=True会在每次取连接时发一个轻量ping包,确认连接还活着。这对于跑在云环境、数据库连接可能被中间网络设备回收的场景特别有用,能避免"Connection has been closed"这种偶发报错。
4.3 高性能异步方案
如果你的项目对并发要求高,可以上SQLAlchemy异步版本。异步版要换一套API,不能和同步版混用。
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession from typing import AsyncGenerator DATABASE_URL = "mysql+asyncmy://root:password@localhost/fastapi_demo?charset=utf8mb4" async_engine = create_async_engine( DATABASE_URL, pool_size=10, max_overflow=20, pool_pre_ping=True, echo=False, ) AsyncSessionLocal = async_sessionmaker( async_engine, class_=AsyncSession, expire_on_commit=False, ) async def get_async_db() -> AsyncGenerator[AsyncSession, None]: async with AsyncSessionLocal() as session: yield session使用异步数据库时,路由函数必须用async def定义,而且所有数据库操作都要加await:
from sqlalchemy import select @router.get("/books/{book_id}") async def get_book(book_id: int, db: AsyncSession = Depends(get_async_db)): result = await db.execute(select(Book).where(Book.id == book_id)) book = result.scalar_one_or_none() if not book: raise HTTPException(status_code=404, detail="Book not found") return book我在实际项目中遇到过一个问题:很多教程里讲result.scalar_one_or_none()取单条结果,但如果你没注意查询可能返回多条,SQLAlchemy会直接抛MultipleResultsFound异常。在设计中,get_book按主键查,永远只有零条或一条,所以用这个方法是安全的。如果你按某个非唯一字段查,要改用scalars().first(),或者手动处理多条结果。
4.4 分页、过滤与索引设计
列表接口必须做分页,否则数据量一大,一次查全表既慢又浪费内存。最基础的方案是limit/offset分页。
@router.get("/books/") def list_books( skip: int = 0, limit: int = 10, db: Session = Depends(get_db), ): books = db.query(Book).order_by(Book.id).offset(skip).limit(limit).all() return books但你得知道limit/offset的软肋:偏移量越大,数据库要扫的行越多。offset=100000时,数据库得先查出前100010行再丢掉前100000行,效率非常低。数据量超过几十万之后,更推荐用游标分页,也叫keyset pagination,核心思路是用上次返回的最后一个id作为查询条件。
@router.get("/books/") def list_books( after_id: int = 0, limit: int = 10, db: Session = Depends(get_db), ): books = db.query(Book).where(Book.id > after_id).order_by(Book.id).limit(limit).all() return books这种分页方式的性能不随页数增加而下降,因为where id > ?配合主键索引可以瞬间定位数据位置。当然它也有局限:无法跳页,只能一页一页往下翻。适合做"加载更多"这种交互形式,不适合做带页号跳转的后台管理系统。
索引的设计同样重要。查询频繁用的条件字段,比如title、author,可以加index=True。但索引不是越多越好,每个索引都会拖慢写入速度,还会占用磁盘空间。我的习惯是:先根据实际查询语句分析where和排序字段,只给高频查询加索引,宁缺毋滥。
5. 实战:纯REST接口从零到一
5.1 需求设计与接口规划
这个部分我们做一个完整的图书管理REST接口,把前面讲的知识点串起来。REST接口的设计核心是:资源用名词复数命名,操作语义通过HTTP方法表达,状态码准确传达结果。
| 方法 | 路径 | 语义 | 成功状态码 | 失败状态码 |
|---|---|---|---|---|
| POST | /books/ | 创建图书 | 201 Created | 422 参数校验失败 |
| GET | /books/ | 获取图书列表(分页) | 200 OK | - |
| GET | /books/{book_id} | 获取图书详情 | 200 OK | 404 不存在 |
| PUT | /books/{book_id} | 更新图书 | 200 OK | 404/422 |
| DELETE | /books/{book_id} | 删除图书 | 204 No Content | 404 |
纯REST接口有个容易忽略的点:创建成功应该返回201,删除成功应该返回204(空响应体),而不是全部返回200。很多后端同学习惯统一返回200然后自己在body里塞一个{"code": 1, "msg": "success"},这在REST语义下是不推荐的。HTTP状态码本身就是表达语义的工具,搞一套自定义code来包装,等于把工具废了又自己造轮子。
5.2 模型与Schema定义
模型层已经在4.1节定义好了,现在定义Pydantic的Schema。这里我遵循一个最佳实践:输入的Schema和输出的Schema分开定义。
from pydantic import BaseModel, ConfigDict, Field from datetime import datetime class BookCreate(BaseModel): title: str = Field(..., min_length=1, max_length=200, description="书名") author: str = Field(..., min_length=1, max_length=100, description="作者") price: float = Field(..., gt=0, description="价格") stock: int = Field(0, ge=0, description="库存") class BookUpdate(BaseModel): title: str | None = Field(None, min_length=1, max_length=200) author: str | None = Field(None, min_length=1, max_length=100) price: float | None = Field(None, gt=0) stock: int | None = Field(None, ge=0) class BookRead(BaseModel): model_config = ConfigDict(from_attributes=True) id: int title: str author: str price: float stock: int created_at: datetimeBookCreate里Field(..., min_length=1)表示必填且长度至少1,前端漏传或传空字符串都会得到422。BookUpdate里所有字段都是可选,这是为了支持部分更新。最关键的BookRead里的ConfigDict(from_attributes=True),它告诉Pydantic可以从SQLAlchemy模型实例直接转换。没有这一行,把ORM对象当作响应返回时会报AttributeError,这是Pydantic v2的写法,v1的老代码写的是class Config: orm_mode = True,网上旧教程会让你踩这个坑。
输入输出分离的理由很实际:客户端的BookCreate不携带id和created_at,因为这是服务端生成的字段,用户传了也不应该被接受;而BookRead则展示完整资源信息。如果共用同一个Schema,很容易出现客户端把不可控字段传上来,或者响应里多出不该出现的内部字段。
5.3 路由与CRUD接口实现
先把路由器和依赖注入接起来。
from fastapi import APIRouter, Depends, HTTPException, status from sqlalchemy.orm import Session from app.schemas.book import BookCreate, BookUpdate, BookRead from app.models.book import Book from app.dependencies import get_db router = APIRouter(prefix="/books", tags=["books"])创建图书的接口,核心是三步:接收Schema、写入数据库、返回带id和created_at的完整对象。
@router.post("/", response_model=BookRead, status_code=status.HTTP_201_CREATED) def create_book(payload: BookCreate, db: Session = Depends(get_db)): book = Book(**payload.model_dump()) db.add(book) db.commit() db.refresh(book) return book这里有几个关键点。payload.model_dump()是Pydantic v2里官方推荐的方法,v1里叫dict(),返回一个普通字典,然后通过Book(**dict)拆包成ORM实例。db.refresh(book)的作用是从数据库重新拉取这行数据,因为id和created_at是数据库生成的,不refresh的话,返回给前端时这两个字段是空的。提交之后session里的对象状态未过期,但自增主键和数据库默认值不会被自动回填,所以refresh这步必须做。
获取列表接口支持分页和标题模糊搜索:
@router.get("/", response_model=list[BookRead]) def list_books( skip: int = 0, limit: int = 10, title: str | None = None, db: Session = Depends(get_db), ): query = db.query(Book) if title: query = query.filter(Book.title.contains(title)) books = query.order_by(Book.id).offset(skip).limit(limit).all() return booksBook.title.contains(title)会生成WHERE title LIKE '%xxx%'的SQL,适合小规模模糊搜索。当数据量大了之后,%xxx%这种写法因为最前面有通配符,数据库没法走索引,查询会退化成全表扫描。这时候要么用全文索引,要么引入专门的搜索引擎,先有个预期,后面才不会措手不及。
更新接口用PUT,这里有个技巧:
@router.put("/{book_id}", response_model=BookRead) def update_book( book_id: int, payload: BookUpdate, db: Session = Depends(get_db), ): book = db.get(Book, book_id) if not book: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Book not found") update_data = payload.model_dump(exclude_unset=True) for field, value in update_data.items(): setattr(book, field, value) db.commit() db.refresh(book) return bookexclude_unset=True这个参数非常实用。它会让Pydantic只返回客户端显式传过的字段,没传的字段值不会被拿出来覆盖数据库里的旧值。这样就能实现"客户端传哪个字段就更新哪个字段,不传的保持原样"。如果漏了这个参数,BookUpdate里没传的字段就会以None覆盖原有数据,造成莫名其妙的数据丢失。
删除接口返回204,No Content表示删除成功但没有响应体。
@router.delete("/{book_id}", status_code=status.HTTP_204_NO_CONTENT) def delete_book(book_id: int, db: Session = Depends(get_db)): book = db.get(Book, book_id) if not book: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Book not found") db.delete(book) db.commit() return None注意204状态码下FastAPI要求响应体为空,所以return None是必须的,如果返回一个dict,可能会收到"Response body is not allowed for 204"的警告或报错。另外,删除操作会直接物理删除行记录,如果业务希望保留历史数据,建议给表加一个is_deleted字段,查询时统一过滤,这种方式叫软删除,很多正规项目都这么做。
5.4 注册路由与统一响应格式的讨论
最后在main.py里注册路由。
from fastapi import FastAPI from app.routers import book app = FastAPI(title="图书管理系统API", version="1.0.0") app.include_router(book.router)关于统一响应格式,我多说两句。网上很多教程教你把所有接口包一层{"code": 0, "message": "success", "data": ...},这种设计在企业内很常见,但纯REST的观点是:状态码就该用HTTP的,成功和失败的信息就该用HTTP状态码和错误体表达,没必要再套一层。我的倾向是:对外公开的API尽量保持纯REST,响应体直接就是资源本身,错误通过HTTP状态码和detail字段表达;如果是对内服务且历史约定就是包一层,那就保持团队一致,不要混着来。这里没有绝对的对错,但一定要统一,混用会让前端对接的人疯掉。
6. 常见问题与排查技巧实录
6.1 问题速查表
这一节我把实战中遇到的高频问题整理成一张速查表,方便大家按图索骥。
| 现象 | 原因 | 解决办法 |
|---|---|---|
启动时报AttributeError: 'BaseQuery' object has no attribute 'filter_by' | SQLAlchemy版本混用 | 检查是否混用了2.0和1.x写法,统一用2.0风格 |
Pydantic报orm_mode相关错误 | 教程基于Pydantic v1 | v2改为ConfigDict(from_attributes=True) |
| 接口返回422而不是自定义的400 | Pydantic校验不通过 | 这是正常行为,422表示语义正确但校验失败 |
| 前端跨域请求在OPTIONS阶段就挂了 | CORS配置不完整 | 检查allow_origins是否包含实际来源,且含OPTIONS方法 |
偶发MySQL Connection is not available | 连接被网络设备回收 | pool_pre_ping=True |
| 页面请求一次后卡死 | 异步路由里用了同步数据库调用 | 改用异步引擎或用def定义路由 |
SQLite报database is locked | 并发写同一sqlite文件 | 开发环境换成PostgreSQL或MySQL |
| 修改代码不生效 | --reload没开或改了没触发 | 确认启动命令带--reload |
6.2 几个印象深刻的踩坑经历
第一个坑是关于缓存失效的。每个请求挂了数据库连接池之后,某个凌晨收到告警,数据库连接数直线上升,最后把连接池打爆导致服务不可用。查下来发现是某些慢查询把连接占用时间拉长了,新请求不断申请新连接,max_overflow=20很快就用满了。这个问题的根源不是连接池配置,而是慢查询。所以连接池参数只是兜底,真正要做的是把响应时间控制在合理范围,同时监控慢查询日志。
第二个坑是异步和同步的混用。有一次在async def路由里,直接调用同步SQLAlchemy的db.query方法查询数据,接口正常运行,但一旦并发高起来,整个事件循环被数据库阻塞,其他接口也跟着卡。排查半天才意识到问题所在:同步数据库操作是阻塞式的,在异步事件循环里执行,相当于把所有请求串行排队了。解决方案是:要么全部走异步引擎,要么把这个路由改成普通的def,FastAPI会自动丢到线程池执行,不会阻塞主循环。这个坑不止我一个人踩,热搜里的"fastapi和sqlalchemy构建高性能web服务"大概率也有不少人在找这个答案。
第三个坑是关于字段默认值的。项目开发阶段用SQLite,created_at字段用Python侧default=datetime.now,一切正常。迁移到MySQL后,发现直接通过SQL插入的数据没有创建时间,报错才发现Python侧默认值在数据库层根本不存在。后来统一改成server_default=func.now(),让数据库自己生成时间,才彻底解决。这个教训告诉我,写ORM模型时,凡是数据库能自己搞定的默认值,就不要依赖应用层。
6.3 生产环境部署要点
部署这块简单说说。直接跑uvicorn app.main:app是开发模式,生产环境需要多进程和进程管理。推荐用gunicorn作为进程管理器,让uvicorn充当ASGI worker:
gunicorn app.main:app -k uvicorn.workers.UvicornWorker -w 4 --bind 0.0.0.0:8000worker数量一般建议CPU核心数*2+1,比如4核机器就开9个worker,这个公式是Gunicorn官方文档推荐的经验值。如果觉得手工维护太麻烦,Dockerfile方案也很成熟:
FROM python:3.11-slim WORKDIR /app COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt COPY . . EXPOSE 8000 CMD ["gunicorn", "app.main:app", "-k", "uvicorn.workers.UvicornWorker", "-w", "4", "--bind", "0.0.0.0:8000"]--no-cache-dir能让镜像小不少,python:3.11-slim比python:3.11体积小很多,部署时要刻意关注镜像体积,特别是服务数量多的时候。还有一个容易被忽略的点:Docker里跑的进程默认是root权限,存在安全风险,正规做法是单独建一个非root用户运行应用。这一点在团队协作时可能不那么被重视,但作为个人项目上线时一定要养成习惯。
6.4 性能优化的几条实测建议
性能优化是"fastapi和sqlalchemy构建高性能web服务"这个热搜词绕不开的话题。我的实测经验集中在四条:第一,尽量用async def定义路由函数。第二,数据库查询尽量精确,只select需要的列,避免select *。第三,热点数据上Redis缓存,特别是那种被频繁读取但很少变化的数据,数据库压力能降两个数量级。第四,接口里的逻辑如果能合并成一条SQL,就不要拆成多次查询往返。
缓存这块举个例子。图书列表接口如果每个用户进来都查一次数据库,数据库压力会很大。用Redis缓存列表数据,设置60秒过期,就能显著降低数据库负载。实现也简单:
import json import redis r = redis.Redis(host="localhost", port=6379, db=0) @router.get("/") def list_books(db: Session = Depends(get_db)): cache_key = "books:list" cached = r.get(cache_key) if cached: return json.loads(cached) books = db.query(Book).order_by(Book.id).limit(10).all() r.setex(cache_key, 60, json.dumps([b.__dict__ for b in books], default=str)) return books注意一个细节:缓存的数据如果是ORM对象,序列化时要处理datetime字段,json.dumps默认不认识datetime对象,要加default=str或者先转换为字符串。这里我展示的是最简写法,生产环境建议引入专门的序列化方案,避免反复踩坑。
在最后再分享一个实践技巧:当你使用依赖注入的get_db模式后,写接口的单元测试会变得很轻松。测试时只需要用依赖覆盖机制,把get_db替换成一个测试用的会话工厂,就能对每个接口做隔离测试,不需要真的连生产数据库。这个操作在FastAPI里只需要一行:
app.dependency_overrides[get_db] = get_test_db这个机制是我个人最喜欢的FastAPI特性之一。它让测试代码极度干净,不用mocking网络请求、不用起真实服务,直接调用路由函数或者用TestClient发请求,数据都落在测试专用的数据库里,跑完就清理。前面我们把项目结构拆得那么清晰,依赖注入又提供了接口隔离的基础,写起测试来顺手很多。
FastAPI这套东西上手之后,写接口的效率确实比传统框架高出一截。从一开始被类型注解的写法搞得有点不适应,到后来习惯了先定义Schema再写路由,整个开发流顺畅了很多。这篇整理了项目从零到上线的完整链路,希望对你有所帮助。