Python FastAPI + SQLAlchemy 入门教程：从零搭建你的第一个 Web 应用

像搭积木一样一步步构建完整的 Web 服务 本文使用的代码文件在开始之前请先创建以下三个核心文件所有代码都将基于它们展开myproject/ ├── main.py # FastAPI 主入口包含所有路由 ├── database.py # 数据库配置引擎、会话、依赖注入 ├── models.py # ORM 模型Category 和 Post └── seed.py # 测试数据填充可选前言作为一名 Python 开发者想要快速搭建一个 Web 应用FastAPI SQLAlchemy 的组合是目前最流行的选择之一。本文将用积木式的方式带你从零开始一步步搭建一个完整的博客文章管理系统。你将学到✅ FastAPI 路由、请求处理、响应✅ SQLAlchemy ORM 模型定义与数据库操作✅ Pydantic 数据校验✅ 依赖注入、中间件、启动事件✅ 完整的项目结构前置知识Python 基础语法、了解 HTTP 协议基本概念第一块积木数据库连接database.py对应文件database.py完整代码见文末1.1 创建数据库引擎fromsqlalchemyimportcreate_engine enginecreate_engine(mysqlpymysql://root:123456localhost:3306/pythondb,echoTrue,# 打印执行的 SQL开发调试用pool_size5,# 连接池大小max_overflow10# 超出连接池后的最大连接数)# 测试连接withengine.connect()asconn:print(连接成功) 关键点连接字符串格式mysqlpymysql://用户名:密码主机:端口/数据库名记得先在 MySQL 中创建对应的数据库pythondb1.2 创建会话工厂和基类fromsqlalchemy.ormimportsessionmaker,declarative_base# 会话工厂用于创建数据库会话SessionLocalsessionmaker(autocommitFalse,autoflushFalse,bindengine)# ORM 基类所有模型都继承它Basedeclarative_base()提示文件中还定义了一个Base1类那是另一种写法示例实际使用时直接用Base即可。1.3 依赖注入获取数据库会话fromfastapiimportDependsdefget_db():每个请求创建独立的数据库会话请求结束后自动关闭dbSessionLocal()try:yielddb# 请求期间使用这个会话finally:db.close()# 防止连接泄漏为什么这样设计每个请求使用独立的会话互不干扰自动管理连接生命周期避免资源泄漏配合 FastAPI 的依赖注入系统使用第二块积木数据模型models.py对应文件models.py完整代码见文末定义数据库表结构就像画施工图纸一样。2.1 分类表CategoryfromsqlalchemyimportColumn,Integer,String,Text,DateTime,ForeignKeyfromsqlalchemy.ormimportrelationshipfromdatetimeimportdatetimefromdatabaseimportBaseclassCategory(Base):文章分类__tablename__categoriesidColumn(Integer,primary_keyTrue,indexTrue)nameColumn(String(50),uniqueTrue,nullableFalse)slugColumn(String(50),uniqueTrue,nullableFalse)# 与 Post 表建立双向关系postsrelationship(Post,back_populatescategory)2.2 文章表PostclassPost(Base):博客文章__tablename__postsidColumn(Integer,primary_keyTrue,indexTrue)titleColumn(String(200),nullableFalse)slugColumn(String(200),uniqueTrue,nullableFalse)summaryColumn(Text,default)contentColumn(Text,nullableFalse)category_idColumn(Integer,ForeignKey(categories.id),nullableFalse)created_atColumn(DateTime,defaultdatetime.utcnow)updated_atColumn(DateTime,defaultdatetime.utcnow,onupdatedatetime.utcnow)# 与 Category 的双向关系categoryrelationship(Category,back_populatesposts) 字段说明字段类型说明idInteger主键自动索引nameString(50)分类名称唯一slugString(200)URL 友好的标识符必须唯一summaryText文章摘要默认为空字符串category_idInteger外键关联 categories 表created_atDateTime创建时间自动填充当前时间updated_atDateTime更新时间每次修改自动更新第三块积木Pydantic 数据校验对应位置main.pyPydantic 是 FastAPI 的数据校验核心自动生成 API 文档自动校验输入数据。3.1 基础模型frompydanticimportBaseModel,EmailStrclassUserBase(BaseModel):username:str# 必填email:EmailStr# 必填自动校验邮箱格式full_name:str|NoneNone# 可选3.2 输入输出分离# 创建用户时的输入模型含密码classUserCreate(UserBase):password:str# 返回用户信息的输出模型不含密码classUserOut(UserBase):id:intapp.post(/users/,response_modelUserOut)asyncdefcreate_user(user:UserCreate):# 响应时自动过滤密码字段return{id:1,**user.model_dump(exclude{password})}✨最佳实践输入和输出模型分开定义避免敏感信息泄露3.3 商品模型示例classItem(BaseModel):name:strage:intNone# 可选字段app.post(/items1)defitems1(item:Item):print(item.model_dump())# 转为字典print(item.model_dump_json())# 转为 JSON 字符串returnitem第四块积木FastAPI 路由main.py对应位置main.py4.1 基础路由fromfastapiimportFastAPI appFastAPI()# GET 请求路径参数app.get(/)asyncdefroot():return{message:Hello World}app.get(/items/{item_id})defread_item(item_id:int,q1:strNone,p2:strp22):return{item_id:item_id,q:q1,p2:p2}# POST 请求接收 JSON Bodyapp.post(/items1)defitems1(item:Item):returnitem4.2 请求头和 CookiefromfastapiimportHeader,Cookieapp.get(/item2)defitem2(user_agent:strHeader(None),session_token:strCookie(None)):return{User-Agent:user_agent,Session-Token:session_token}4.3 重定向和状态码fromfastapi.responsesimportRedirectResponse,JSONResponsefromfastapiimportHTTPException# 重定向app.get(/redirect)defredirect():returnRedirectResponse(url/item2/)# 抛出 HTTP 异常app.get(/items1/{item_id})defread_item(item_id:int):ifitem_id42:raiseHTTPException(status_code404,detailItem not found)return{item_id:item_id}# 自定义响应头app.get(/default_header)defdefault_header():returnJSONResponse(content{item_id:1000,body:10001},headers{X-Custom-Header:custom-value})4.4 表单数据处理fromfastapiimportFormapp.post(/login/)asyncdeflogin(username:strForm(),# 必填password:strForm(),# 必填):print(f{password})# 调试输出return{username:username}app.post(/items/)asyncdefcreate_item(name:strForm(...),# ... 表示必填description:str|NoneForm(None),price:floatForm(...,gt0),# 必须大于 0):return{name:name,description:description,price:price}第五块积木路由分组APIRouter对应位置main.py当路由越来越多时用APIRouter进行模块化管理。fromfastapiimportAPIRouter routerAPIRouter(prefix/posts,# 所有路由自动加上 /posts 前缀tags[文章]# Swagger 文档分组)router.get(/{post_id},namepost_detail)defpost_detail(post_id:int):文章详情return{post_id:post_id}# 挂载到主应用app.include_router(router)第六块积木启动事件 - 自动建表对应位置main.py应用启动时自动创建数据库表开发阶段非常方便。fromdatabaseimportengine,BasefrommodelsimportCategory,Post# 确保模型被导入app.on_event(startup)defon_startup():应用启动时自动创建数据库表仅开发使用Base.metadata.create_all(bindengine)print(✅ 数据库表创建成功)⚠️注意如果表已存在create_all不会重复创建只会创建不存在的表。第七块积木数据填充seed.py⚠️注意seed.py文件在提供的代码中未给出完整实现但在main.py中被引用。# main.py 中的使用importseedapp.post(/insert_data)asyncdefinsert_data():seed.exec()# 执行数据填充returnsuccess...你需要自己创建seed.py文件示例内容# seed.py - 你需要自行创建fromdatabaseimportSessionLocalfrommodelsimportCategory,Postdefexec():dbSessionLocal()try:# 创建分类catCategory(namePython,slugpython)db.add(cat)db.commit()# 创建文章postPost(titleFastAPI 入门,slugfastapi-intro,content这是内容...,category_idcat.id)db.add(post)db.commit()print(✅ 测试数据填充成功)finally:db.close()第八块积木完整代码汇总 database.py完整fromsqlalchemyimportcreate_enginefromsqlalchemy.ormimportsessionmaker,declarative_basefromfastapiimportDepends# MySQL连接字符串enginecreate_engine(mysqlpymysql://root:123456localhost:3306/pythondb,echoTrue,pool_size5,max_overflow10)# 测试连接withengine.connect()asconn:print(连接成功)# 会话工厂SessionLocalsessionmaker(autocommitFalse,autoflushFalse,bindengine)# ORM 基类Basedeclarative_base()# 依赖注入defget_db():dbSessionLocal()try:yielddbfinally:db.close() models.py完整fromsqlalchemyimportColumn,Integer,String,Text,DateTime,ForeignKeyfromsqlalchemy.ormimportrelationshipfromdatetimeimportdatetimefromdatabaseimportBaseclassCategory(Base):__tablename__categoriesidColumn(Integer,primary_keyTrue,indexTrue)nameColumn(String(50),uniqueTrue,nullableFalse)slugColumn(String(50),uniqueTrue,nullableFalse)postsrelationship(Post,back_populatescategory)classPost(Base):__tablename__postsidColumn(Integer,primary_keyTrue,indexTrue)titleColumn(String(200),nullableFalse)slugColumn(String(200),uniqueTrue,nullableFalse)summaryColumn(Text,default)contentColumn(Text,nullableFalse)category_idColumn(Integer,ForeignKey(categories.id),nullableFalse)created_atColumn(DateTime,defaultdatetime.utcnow)updated_atColumn(DateTime,defaultdatetime.utcnow,onupdatedatetime.utcnow)categoryrelationship(Category,back_populatesposts) main.py核心部分fromfastapiimportFastAPI,Depends,Header,Cookie,Form,HTTPExceptionfromfastapi.responsesimportRedirectResponse,JSONResponsefromfastapiimportAPIRouterfrompydanticimportBaseModel,EmailStrimportdatabase,models,seed appFastAPI()# ... 所有路由定义参考第四、五部分...app.on_event(startup)defon_startup():models.Base.metadata.create_all(binddatabase.engine)# 挂载路由app.include_router(router)快速启动# 1. 安装依赖pipinstallfastapi uvicorn sqlalchemy pymysql pydantic# 2. 创建数据库MySQLmysql-uroot-pCREATE DATABASE pythondb;# 3. 启动服务热重载模式uvicorn main:app--reload# 4. 访问交互式 API 文档# http://127.0.0.1:8000/docs总结积木清单积木文件作用状态 数据库引擎database.py连接数据库✅ 已提供 会话工厂database.py管理数据库会话✅ 已提供 依赖注入database.py获取数据库会话✅ 已提供 ORM 模型models.py定义表结构✅ 已提供 Pydantic 模型main.py数据校验✅ 已提供 路由定义main.pyAPI 端点✅ 已提供 路由分组main.py模块化管理✅ 已提供 启动事件main.py自动建表✅ 已提供 数据填充seed.py测试数据⚠️ 需自行创建⚠️ 注意事项seed.py 文件未提供main.py中引用了import seed和seed.exec()你需要自行创建该文件或注释掉相关路由。修改数据库配置将database.py中的连接字符串改为你自己的 MySQL 配置。生产环境注意关闭echoTrue不要在生产环境使用create_all()使用环境变量管理敏感信息下一步可以学什么关联查询学习joinedload预加载关联数据分页查询实现offset/limit分页中间件添加日志、CORS、认证异步支持使用async/await处理高并发单元测试用TestClient测试 API记住编程就像搭积木先把基础模块理解透彻再逐步组合成复杂的系统。本文的每个代码块都是独立的积木你可以随时取用如果你在实践过程中遇到问题欢迎在评论区留言交流