FastAPI 终极实战ORM 数据库、RESTful 设计、中间件与依赖注入这是 FastAPI 系列的最后一篇。你将学会使用 SQLAlchemy ORM 操作数据库、设计符合 RESTful 规范的 API、理解 HTTP 状态码的语义、模块化路由、依赖注入复用逻辑、中间件统一处理请求、解决跨域问题、以及提供静态文件服务。最终完成一个完整的图书管理 API 项目。目录ORM 基础SQLAlchemy 集成1.1 核心组件Engine、Base、Session1.2 定义模型1.3 创建表1.4 增删改查操作RESTful API 设计原则2.1 核心原则URL 描述资源HTTP 方法描述操作2.2 示例学生资源的增删改查HTTP 状态码详解与高频对比3.1 2xx 成功3.2 3xx 重定向3.3 4xx 客户端错误3.4 5xx 服务器错误3.5 易混淆对比400 vs 422、401 vs 403、307 vs 308、500 vs 503子路由模块化依赖注入Depends中间件Middleware6.1 作用与实现结构6.2 常见应用日志、耗时统计、认证6.3 多中间件顺序跨域CORS配置静态资源服务综合实战图书管理 API完整代码总结与系列回顾1. ORM 基础SQLAlchemy 集成SQLAlchemy 是 Python 最流行的 ORM对象关系映射库它将数据库表映射为 Python 类将行映射为对象将列映射为属性。1.1 核心组件组件说明Engine数据库连接的核心管理连接池Base模型基类所有模型继承它Session数据库操作会话类似“工作单元”1.2 定义模型fromsqlalchemyimportcreate_engine,Column,Integer,Stringfromsqlalchemy.ext.declarativeimportdeclarative_basefromsqlalchemy.ormimportsessionmaker# 创建引擎以 SQLite 为例enginecreate_engine(sqlite:///./test.db,echoTrue)# 模型基类Basedeclarative_base()classUser(Base):__tablename__usersidColumn(Integer,primary_keyTrue,indexTrue)nameColumn(String(50),nullableFalse)ageColumn(Integer)1.3 创建表Base.metadata.create_all(bindengine)1.4 增删改查操作SessionLocalsessionmaker(autocommitFalse,autoflushFalse,bindengine)dbSessionLocal()# 新增单条userUser(name张三,age20)db.add(user)db.commit()db.refresh(user)# 新增多条db.add_all([User(name李四,age22),User(name王五,age25)])db.commit()# 查询all_usersdb.query(User).all()# 所有adultsdb.query(User).filter(User.age18).all()# 条件过滤pageddb.query(User).offset(0).limit(10).all()# 分页# 修改userdb.query(User).filter(User.name张三).first()ifuser:user.age21db.commit()# 删除db.query(User).filter(User.name李四).delete()db.commit()⚠️注意所有写操作必须执行db.commit()才会写入数据库。操作完成后建议关闭 sessiondb.close()可使用try/finally确保释放。2. RESTful API 设计原则RESTful 是一种 API 设计规范旨在提高可读性与可维护性。核心原则说明示例URL 描述资源使用名词复数避免动词/users而不是/getUserHTTP 方法描述操作GET 查询、POST 新增、PUT 修改、DELETE 删除GET /users/1获取 id1 的用户状态码语义化用状态码表达结果类型200 OK, 201 Created, 404 Not FoundRESTful 示例请求含义GET /students获取所有学生POST /students新增一个学生PUT /students/{id}修改指定学生DELETE /students/{id}删除指定学生3. HTTP 状态码详解与高频对比3.1 2xx 成功状态码含义使用场景200 OK请求成功GET、PUT、DELETE 成功201 Created创建成功POST 创建资源成功后返回通常带 Location 头204 No Content成功但无返回体DELETE 成功后常用3.2 3xx 重定向状态码含义特点307 Temporary Redirect临时重定向保持请求方法不变308 Permanent Redirect永久重定向保持请求方法不变3.3 4xx 客户端错误状态码含义典型原因400 Bad Request请求格式错误或缺少必要参数缺少查询参数如/search缺少 keyword401 Unauthorized未登录未提供 token 或 token 无效403 Forbidden已登录但无权限普通用户尝试删除其他用户数据404 Not Found资源不存在访问/users/99999但该 id 不存在422 Unprocessable Entity数据校验失败格式正确但语义错误邮箱格式错误、年龄超出范围3.4 5xx 服务器错误状态码含义典型原因500 Internal Server Error服务器内部错误代码抛出未捕获异常、数据库崩溃503 Service Unavailable服务不可用过载、熔断、维护中3.5 易混淆对比对比区分要点400 vs 422400请求格式错误如 JSON 解析失败422格式正确但字段值不符合业务规则401 vs 403401未认证未登录403已认证但无权限307 vs 308307 临时308 永久两者都保留请求方法区别于 301/302500 vs 503500 内部错误代码 bug503 服务不可用过载、维护4. 子路由模块化当项目变大时将所有路由写在main.py会导致臃肿难维护。使用APIRouter拆分模块。# routers/students.pyfromfastapiimportAPIRouter routerAPIRouter(prefix/students,tags[学生管理])router.get(/)asyncdefget_students():return[{name:张三},{name:李四}]router.post(/)asyncdefcreate_student():return{msg:创建成功}在主应用中引入# main.pyfromfastapiimportFastAPIfromroutersimportstudents,teachers appFastAPI()app.include_router(students.router)app.include_router(teachers.router)5. 依赖注入Depends依赖注入用于抽离公共逻辑如分页参数、认证、数据库会话。FastAPI 的Depends会自动解析并注入。fromfastapiimportDependsdefpagination_params(skip:int0,limit:int10):return{skip:skip,limit:limit}app.get(/items)asyncdefget_items(params:dictDepends(pagination_params)):returnparams依赖函数可以是普通函数或异步函数甚至可以依赖其他依赖。6. 中间件Middleware中间件在每个请求前和响应后执行可用于日志、耗时统计、权限校验、统一响应头等。6.1 作用与实现结构fromfastapiimportRequestimporttimeapp.middleware(http)asyncdefadd_process_time_header(request:Request,call_next):start_timetime.time()responseawaitcall_next(request)# 继续处理请求process_timetime.time()-start_time response.headers[X-Process-Time]str(process_time)returnresponse6.2 常见应用耗时中间件统计接口处理时间认证中间件校验 token支持白名单路径日志中间件记录请求方法、路径、状态码6.3 多中间件顺序中间件按添加的顺序执行请求时从外到内响应时从内到外。例如app.add_middleware(MiddlewareA)# 先添加的外层app.add_middleware(MiddlewareB)# 后添加的内层7. 跨域CORS配置前后端分离时前端域名与后端不同会产生跨域问题。FastAPI 内置CORSMiddleware。fromfastapi.middleware.corsimportCORSMiddleware app.add_middleware(CORSMiddleware,allow_origins[*],# 允许所有来源生产环境应指定具体域名allow_credentialsTrue,# 允许携带 Cookieallow_methods[*],# 允许所有 HTTP 方法allow_headers[*],# 允许所有请求头)8. 静态资源服务使用StaticFiles挂载静态目录如图片、CSS、JS。fromfastapi.staticfilesimportStaticFiles app.mount(/static,StaticFiles(directorystatic),namestatic)访问http://localhost:8000/static/logo.png即可获取文件。自定义缓存控制继承StaticFiles并重写file_response方法添加Cache-Control头。9. 综合实战图书管理 API完整代码下面结合 ORM、RESTful、子路由、依赖注入、中间件、CORS 等实现完整的图书管理 API。# models.pyfromsqlalchemyimportcreate_engine,Column,Integer,Stringfromsqlalchemy.ext.declarativeimportdeclarative_basefromsqlalchemy.ormimportsessionmaker SQLALCHEMY_DATABASE_URLsqlite:///./books.dbenginecreate_engine(SQLALCHEMY_DATABASE_URL,connect_args{check_same_thread:False})SessionLocalsessionmaker(autocommitFalse,autoflushFalse,bindengine)Basedeclarative_base()classBook(Base):__tablename__booksidColumn(Integer,primary_keyTrue,indexTrue)titleColumn(String,indexTrue)authorColumn(String)yearColumn(Integer)Base.metadata.create_all(bindengine)# crud.py (依赖数据库会话)defget_book(db,book_id:int):returndb.query(Book).filter(Book.idbook_id).first()defget_books(db,skip:int0,limit:int10):returndb.query(Book).offset(skip).limit(limit).all()defcreate_book(db,title:str,author:str,year:int):bookBook(titletitle,authorauthor,yearyear)db.add(book)db.commit()db.refresh(book)returnbookdefdelete_book(db,book_id:int):bookget_book(db,book_id)ifbook:db.delete(book)db.commit()returnbook# dependencies.pyfromfastapiimportDependsfrom.modelsimportSessionLocaldefget_db():dbSessionLocal()try:yielddbfinally:db.close()defpagination(skip:int0,limit:int10):return{skip:skip,limit:limit}# routers/books.pyfromfastapiimportAPIRouter,Depends,HTTPExceptionfromsqlalchemy.ormimportSessionfrom..importcrud,dependenciesfrom..modelsimportBook routerAPIRouter(prefix/books,tags[图书管理])router.get(/)asyncdeflist_books(pagination:dictDepends(dependencies.pagination),db:SessionDepends(dependencies.get_db)):bookscrud.get_books(db,skippagination[skip],limitpagination[limit])returnbooksrouter.post(/,status_code201)asyncdefadd_book(title:str,author:str,year:int,db:SessionDepends(dependencies.get_db)):returncrud.create_book(db,title,author,year)router.get(/{book_id})asyncdefget_book(book_id:int,db:SessionDepends(dependencies.get_db)):bookcrud.get_book(db,book_id)ifnotbook:raiseHTTPException(status_code404,detail图书不存在)returnbookrouter.delete(/{book_id})asyncdefremove_book(book_id:int,db:SessionDepends(dependencies.get_db)):bookcrud.delete_book(db,book_id)ifnotbook:raiseHTTPException(status_code404,detail图书不存在)return{msg:删除成功}# main.pyfromfastapiimportFastAPIfromfastapi.middleware.corsimportCORSMiddlewarefromfastapi.staticfilesimportStaticFilesfromroutersimportbooks appFastAPI(title图书管理API)# CORSapp.add_middleware(CORSMiddleware,allow_origins[*],allow_credentialsTrue,allow_methods[*],allow_headers[*],)# 静态文件app.mount(/static,StaticFiles(directorystatic),namestatic)# 子路由app.include_router(books.router)# 根路径app.get(/)defroot():return{message:图书管理API已启动}启动后访问/docs即可测试所有接口。10. 总结与系列回顾本系列三篇文章涵盖了 FastAPI 从入门到实战的全路径篇目核心内容第一篇HTTP 协议、路由、参数、异常、异步、ASGI、Uvicorn第二篇Pydantic 模型、文件上传、表单、响应模型、自定义校验第三篇本文ORM、RESTful、状态码、子路由、依赖注入、中间件、CORS、静态文件后续你可以将项目部署到云服务器使用 Gunicorn Uvicorn workers添加身份认证JWT集成 Redis 做缓存编写单元测试使用 Docker 容器化FastAPI 是现代 Python Web 开发的不二之选。希望这个系列能帮助你写出专业、高效的 API 服务。