FastAPI + 异步 SQLAlchemy 实战：从零搭建图书管理 CRUD 项目

前言

本篇将从零开始，带你搭建一个完整的异步图书管理 CRUD 项目，覆盖环境搭建、数据库连接、模型定义、12 种核心接口实现。献给和博主一样刚踏入SQLAlchemy的新手小白们。

注意：本文基础知识较多，不需要的大佬可直接跳到具体操作环节

一、基础知识回顾

1.什么是FastAPI

专业版：
FastAPI 是一款基于 Python 类型提示、异步原生支持、高性能、现代化的 Web 后端框架，遵循 OpenAPI 与 JSON Schema 规范，用于快速构建 RESTful API 接口服务；底层基于 Starlette（异步 Web 核心）和 Pydantic（数据校验与类型解析），具备自动接口文档、请求数据自动校验、路径参数解析、依赖注入、生命周期管理等企业级特性，支持同步 / 异步两种编程范式，适配 Python 3.8 及以上版本。
破解版：
你可以把 FastAPI 理解成专门用来写后端接口的 Python 专用工具框架：不用自己从零处理网络请求、参数校验、接口文档、异步并发，它全都帮你封装好了；你只需要写业务逻辑，就能快速写出给前端、小程序、第三方调用的后端接口，而且自带可视化接口文档，运行速度还特别快。
同行：
1.Django
全能型全栈框架，一体化解决方案框架。它是自带全屋家具的精装房，后台、登录、数据库、安全防护全都配齐，适合做大型完整系统，但笨重、不适合单纯写高性能接口。
2.Flask
极简微内核框架，高度可定制化轻量化框架。它是只有毛坯地基的空地，只给最基础架子，想要数据库、文档、校验都要自己额外装插件，小巧自由，但做大项目容易乱。

2.什么是 SQLAlchemy

专业版：
SQLAlchemy 是 Python 生态中最成熟、最强大、企业级标准的 ORM 框架，同时提供 ORM（对象关系映射） 与 Core（原生 SQL 构建工具） 两套核心体系，支持所有主流数据库（MySQL、PostgreSQL、SQLite 等），提供类型安全、事务管理、连接池、关系映射、级联操作等完整数据库操作能力，是 Python 后端访问数据库的事实工业标准。
破解版：SQLAlchemy 就是Python 操作数据库的 “超级翻译官”。你不用写复杂的 SQL 语句，只用写 Python 对象和方法，它自动帮你翻译成 SQL 去执行。

3.同步和异步

同步 SQLAlchemy（传统版）：
基于 同步阻塞 I/O 模型，执行数据库操作时，程序必须等待数据库返回结果才能继续执行后续代码，同一时间只能处理一个任务，属于线性执行模式。
异步 SQLAlchemy（现代版）：
基于 ASGI 异步非阻塞 I/O 模型，使用 async/await 语法，数据库操作期间不会阻塞事件循环，程序可以去处理其他请求，等数据库返回后再回来继续执行，极大提升并发能力。

# 同步引擎fromsqlalchemyimportcreate_enginefromsqlalchemy.ormimportsessionmaker engine=create_engine("mysql+aiomysql://root:password@localhost:3306/数据库名")# 创建同步会话工厂：绑定数据库引擎，用于生产普通同步数据库会话SessionLocal=sessionmaker(bind=engine)# 操作数据库（普通函数）defget_book():db=SessionLocal()res=db.query(Book).all()db.close()returnres#特点：普通函数、无 async/await，执行数据库操作时程序阻塞等待。# 异步引擎fromsqlalchemy.ext.asyncioimportcreate_async_engine,AsyncSessionfromsqlalchemy.ormimportsessionmaker async_engine=create_async_engine("mysql+aiomysql://root:password@localhost:3306/数据库名")# 创建异步会话工厂：绑定异步引擎，指定使用异步会话类 AsyncSessionAsyncSessionLocal=sessionmaker(bind=async_engine,class_=AsyncSession)# 操作数据库（异步函数）asyncdefget_book():asyncwithAsyncSessionLocal()asdb:#await 挂起当前协程，等待数据库返回结果，不阻塞事件循环。res=awaitdb.execute(select(Book))returnres.scalars().all()

二、项目搭建完整流程

1.项目结构规划

我们采用清晰的分层结构，方便维护：

FastAPIProject/ ├── main.py # 项目入口，编写所有接口 ├── database.py # 数据库异步连接配置 └── models.py # ORM 模型定义（数据表）

2.安装依赖

打开 PyCharm 终端，执行安装命令：

pipinstallfastapi uvicorn sqlalchemy[asyncio]aiomysql

3.配置异步数据库连接（database.py）

这是异步项目的核心文件，负责创建异步引擎和数据库会话：

fromsqlalchemy.ext.asyncioimportcreate_async_engine,AsyncSessionfromsqlalchemy.ormimportsessionmaker# 异步数据库链接 格式：mysql+异步驱动://用户名:密码@主机:端口/数据库名ASYNC_DB_URL="mysql+aiomysql://root:你的密码@localhost:3306/你的库名?charset=utf8mb4"# 创建异步数据库引擎async_engine=create_async_engine(ASYNC_DB_URL,echo=False)# 创建异步会话工厂（用于获取数据库连接）AsyncSessionLocal=sessionmaker(bind=async_engine,class_=AsyncSession,expire_on_commit=False)# 依赖项：给接口提供数据库连接asyncdefget_db():asyncwithAsyncSessionLocal()assession:yieldsession

注意：不要害怕，这一整段就是行业标准固定写法，几乎是写 FastAPI 后端人人都复制粘贴的死模板。99% 的 FastAPI + 异步 MySQL 项目直接照搬就能用，不用改内部逻辑。（记得修改自己的的数据库账号和密码哦）

4.定义 ORM 模型（models.py）

ORM 的作用：用 Python 类定义数据表，无需手写 SQL 建表语句。
我们定义两个类：
Base：基类，统一管理创建 / 更新时间（这样我们就不用每个类都定义时间字段了）
Book：图书数据表，包含书名、作者、价格等字段

fromdatetimeimportdatetimefromsqlalchemyimportDateTime,String,Float,funcfromsqlalchemy.ormimportDeclarativeBase,Mapped,mapped_column# 公共基类：所有模型都继承它，自动拥有时间字段classBase(DeclarativeBase):create_time:Mapped[datetime]=mapped_column(DateTime,insert_default=func.now(),comment="创建时间")update_time:Mapped[datetime]=mapped_column(DateTime,insert_default=func.now(),onupdate=func.now(),comment="修改时间")# 图书模型（对应数据库的 book 表）classBook(Base):__tablename__="book"# 数据表名# 字段定义id:Mapped[int]=mapped_column(primary_key=True,autoincrement=True,comment="书籍ID")bookname:Mapped[str]=mapped_column(String(255),comment="书名")author:Mapped[str]=mapped_column(String(255),comment="作者")price:Mapped[float]=mapped_column(Float,comment="价格")publisher:Mapped[str]=mapped_column(String(255),comment="出版社")

详细注释：
1.Base(DeclarativeBase)声明 ORM 模型基类，所有表模型都要继承它（DeclarativeBase），才能被映射成数据库表。
2.mapped_column 就是：把 Python 变量 → 变成数据库表的一列！没有它，就没有字段（左边 Mapped [str] → Python 看的）！它可以定义数据库列的规则，常用参数：

mapped_column(String(255),# 数据库类型（必填）comment="出版社",# 注释nullable=False,# 是否允许为空default="未知",# 默认值unique=True# 是否唯一)

三、核心业务实现（main.py）

1. 项目初始化 + 自动建表

新版 FastAPI 使用 lifespan 替代过时的 on_event，项目启动时自动创建数据表：

fromfastapiimportFastAPI,Depends,HTTPExceptionfromsqlalchemy.ext.asyncioimportAsyncSessionfromsqlalchemyimportselect,funcfromtypingimportOptionalfromcontextlibimportasynccontextmanagerfrompydanticimportBaseModel# 导入自定义模块fromdatabaseimportget_db,async_enginefrommodelsimportBase,Book# 生命周期管理：启动建表，关闭释放连接@asynccontextmanagerasyncdeflifespan(_:FastAPI):# lifespan = 项目一生的钩子函数asyncwithasync_engine.begin()asconn:#开启一个异步数据库连接awaitconn.run_sync(Base.metadata.create_all)# 自动建表yield# 这里开始：项目正常运行，接收请求awaitasync_engine.dispose()# 释放数据库连接池# 创建 FastAPI 实例app=FastAPI(title="异步图书管理系统",lifespan=lifespan)

注意：启动建表的函数依旧可以当成模板（钩子函数 = 系统在特定时间点，自动喊你运行的函数）

2.12 种核心接口实现

按照需求，实现查询、新增、修改、删除、分页、统计、模糊查询等功能：

（1）查询类接口

execute(select(…))

# 1. 获取所有图书@app.get("/books")asyncdefget_all_books(db:AsyncSession=Depends(get_db)):""" Depends(get_db)：自动获取数据库连接 不用你手动开连接、关连接 全部自动管理 最终得到一个叫 db 的对象，用来操作数据库 """result=awaitdb.execute(select(Book))books=result.scalars().all()return{"data":books}

详细注释：

1. result
   类型：AsyncResult（SQLAlchemy 2.0 异步结果对象）
   本质：数据库查询结果集的封装容器，包含原始行数据、元数据、游标状态，内部以行元组形式存储。
   不能直接序列化、不能直接遍历为对象列表
   2.scalars()
   提取每行的第一个标量值，将 Result → ScalarResult
   效果：(Book对象,) → Book对象（去掉外层元组）
   3.await
   只用于：【需要等待 I/O 操作】的代码I/O = 和数据库、网络、文件打交道的操作），所以以上代码，只有db.execute(…)（发送 SQL 给 MySQL）需要添加await

execute(select(…).limit(num))

# 2. 获取第一本图书@app.get("/books/first")asyncdefget_first_book(db:AsyncSession=Depends(get_db)):result=awaitdb.execute(select(Book).limit(1))book=result.scalar_one_or_none()ifnotbook:raiseHTTPException(status_code=404,detail="暂无数据")return{"data":book}

详细注释：
scalar_one_or_none()：有且仅有一条数据 → 返回 Book 对象，无数据 / 多条数据 → 返回 None

get()

# 3. 根据主键 get() 查询单本图书@app.get("/books/get/{book_id}")asyncdefget_book_by_id(book_id:int,db:AsyncSession=Depends(get_db)):book=awaitdb.get(Book,book_id)ifnotbook:raiseHTTPException(status_code=404,detail="书籍不存在")return{"data":book}

详细注释：
get()：根据主键查询的快捷方法

# 4. 路径参数查询图书@app.get("/books/{book_id}")asyncdefget_book_by_path(book_id:int,db:AsyncSession=Depends(get_db)):result=awaitdb.execute(select(Book).where(Book.id==book_id))return{"data":result.scalar_one_or_none()}

# 5. 条件查询：价格≥200@app.get("/books/price/ge200")asyncdefget_books_price_ge200(db:AsyncSession=Depends(get_db)):result=awaitdb.execute(select(Book).where(Book.price>=200))return{"data":result.scalars().all()}

# 6. 模糊查询：作者以周开头@app.get("/books/author/zhou")asyncdefget_books_author_zhou(db:AsyncSession=Depends(get_db)):result=awaitdb.execute(select(Book).where(Book.author.like("周%")))return{"data":result.scalars().all()}

where(条件1, 条件2, 条件3)

# 7. 多条件查询：作者周开头 + 价格>100@app.get("/books/zhou/price/gt100")asyncdefget_books_zhou_and_price(db:AsyncSession=Depends(get_db)):result=awaitdb.execute(select(Book).where(Book.author.like("周%"),Book.price>100))return{"data":result.scalars().all()}

# 8. 统计查询：总数 + 平均价格@app.get("/books/stat")asyncdefget_books_stat(db:AsyncSession=Depends(get_db)):count=awaitdb.scalar(select(func.count(Book.id)))avg_price=awaitdb.scalar(select(func.avg(Book.price)))return{"总数":count,"平均价格":round(avg_priceor0,2)}

详细注释：
返回 一条数据 / 一个数字 → 用 scalar
返回 多条数据（列表）→ 用 execute + scalars ().all ()

# 9. 分页查询@app.get("/books/page")asyncdefget_books_page(page:int=1,size:int=10,db:AsyncSession=Depends(get_db)):#计算偏移量offset=(page-1)*size result=awaitdb.execute(select(Book).limit(size).offset(offset))return{"page":page,"data":result.scalars().all()}

（2）增删改类接口

# Pydantic 模型：校验前端传入数据classBookCreate(BaseModel):bookname:strauthor:strprice:floatpublisher:strclassBookUpdate(BaseModel):bookname:Optional[str]=Noneauthor:Optional[str]=Noneprice:Optional[float]=Nonepublisher:Optional[str]=None

详细注释：
作用：
校验前端传入的数据（比如新增用户时，name 必须是字符串，salary 可以为空）
统一接口请求 / 响应的格式，避免前后端格式不一致
和 ORM 模型解耦：ORM 模型是和数据库绑定的，而 schema 是给接口用的，两者职责分离

# 10. 新增图书@app.post("/books",status_code=201)asyncdefcreate_book(book_in:BookCreate,db:AsyncSession=Depends(get_db)):new_book=Book(**book_in.dict())#加入会话（session），没真正存进数据库，只是暂存db.add(new_book)#真正提交到数据库awaitdb.commit()#从数据库重新读取最新数据awaitdb.refresh(new_book)return{"msg":"添加成功","data":new_book}

详细注释：
1.book_in.dict():把对象 → 变成字典
2.**:把字典 “拆开来” 传给函数 / 类,拆成关键字参数
3.BookCreate和Book并非一一对应，User：数据库模型（对应数据库真实表），UserRequest：请求 Schema（前端传过来的数据模板），两边字段数量不一样、名字可以不一样、用途完全不一样，不用一模一样照搬。
4.db.refresh(new_book)：新增数据后，数据库里已经有 id 了，但你代码里的 new_book 对象，身上还是没有 id,因为id是自增的，BookCreat里没有定义id，同理，数据库自动默认的字段，create_time、update_time等，也需要refresh()。

# 11. 修改图书@app.put("/books/{book_id}")asyncdefupdate_book(book_id:int,book_in:BookUpdate,db:AsyncSession=Depends(get_db)):book=awaitdb.get(Book,book_id)ifnotbook:raiseHTTPException(status_code=404)fork,vinbook_in.dict(exclude_unset=True).items():setattr(book,k,v)awaitdb.commit()return{"msg":"修改成功"}

详细注释：
setattr(book, k, v)：给 book 对象的字段赋值
book_in.dict(exclude_unset=True)：把前端传过来的数据转成字典
exclude_unset=True：只取前端真正传过来的字段，没传的字段忽略不改

# 12. 删除图书@app.delete("/books/{book_id}")asyncdefdelete_book(book_id:int,db:AsyncSession=Depends(get_db)):book=awaitdb.get(Book,book_id)ifnotbook:raiseHTTPException(status_code=404)awaitdb.delete(book)awaitdb.commit()return{"msg":"删除成功"}

运行项目

if__name__=="__main__":importuvicorn uvicorn.run("main:app",host="127.0.0.1",port=8000,reload=True)

四、项目运行与测试

1. 启动项目
   直接运行 main.py，控制台显示服务启动成功即可。

2. 可视化接口测试
   FastAPI 自带自动生成的 API 文档，访问地址：

http://127.0.0.1:8000/docs

五、项目开发思路总结

1. 整体开发流程

（1）规划结构：拆分入口、数据库配置、模型文件
（2）环境准备：安装异步依赖
（3）数据库配置：编写异步连接代码
（4）模型定义：用 ORM 映射数据表
（5）接口开发：按照需求实现增删改查
（6）测试运行：使用自动文档验证功能

2. 异步开发核心规则

(1)函数必须加async声明为异步函数
(2)所有IO操作（数据库查询、提交）必须加 await
(3)使用AsyncSession替代同步的 Session
(4)使用异步数据库驱动（aiomysql）