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 enginecreate_engine(mysqlaiomysql://root:passwordlocalhost:3306/数据库名)# 创建同步会话工厂绑定数据库引擎用于生产普通同步数据库会话SessionLocalsessionmaker(bindengine)# 操作数据库普通函数defget_book():dbSessionLocal()resdb.query(Book).all()db.close()returnres#特点普通函数、无 async/await执行数据库操作时程序阻塞等待。# 异步引擎fromsqlalchemy.ext.asyncioimportcreate_async_engine,AsyncSessionfromsqlalchemy.ormimportsessionmaker async_enginecreate_async_engine(mysqlaiomysql://root:passwordlocalhost:3306/数据库名)# 创建异步会话工厂绑定异步引擎指定使用异步会话类 AsyncSessionAsyncSessionLocalsessionmaker(bindasync_engine,class_AsyncSession)# 操作数据库异步函数asyncdefget_book():asyncwithAsyncSessionLocal()asdb:#await 挂起当前协程等待数据库返回结果不阻塞事件循环。resawaitdb.execute(select(Book))returnres.scalars().all()二、项目搭建完整流程1.项目结构规划我们采用清晰的分层结构方便维护FastAPIProject/ ├── main.py # 项目入口编写所有接口 ├── database.py # 数据库异步连接配置 └── models.py # ORM 模型定义数据表2.安装依赖打开 PyCharm 终端执行安装命令pipinstallfastapi uvicorn sqlalchemy[asyncio]aiomysql3.配置异步数据库连接database.py这是异步项目的核心文件负责创建异步引擎和数据库会话fromsqlalchemy.ext.asyncioimportcreate_async_engine,AsyncSessionfromsqlalchemy.ormimportsessionmaker# 异步数据库链接 格式mysql异步驱动://用户名:密码主机:端口/数据库名ASYNC_DB_URLmysqlaiomysql://root:你的密码localhost:3306/你的库名?charsetutf8mb4# 创建异步数据库引擎async_enginecreate_async_engine(ASYNC_DB_URL,echoFalse)# 创建异步会话工厂用于获取数据库连接AsyncSessionLocalsessionmaker(bindasync_engine,class_AsyncSession,expire_on_commitFalse)# 依赖项给接口提供数据库连接asyncdefget_db():asyncwithAsyncSessionLocal()assession:yieldsession注意不要害怕这一整段就是行业标准固定写法几乎是写 FastAPI 后端人人都复制粘贴的死模板。99% 的 FastAPI 异步 MySQL 项目直接照搬就能用不用改内部逻辑。记得修改自己的的数据库账号和密码哦4.定义 ORM 模型models.pyORM 的作用用 Python 类定义数据表无需手写 SQL 建表语句。我们定义两个类Base基类统一管理创建 / 更新时间这样我们就不用每个类都定义时间字段了Book图书数据表包含书名、作者、价格等字段fromdatetimeimportdatetimefromsqlalchemyimportDateTime,String,Float,funcfromsqlalchemy.ormimportDeclarativeBase,Mapped,mapped_column# 公共基类所有模型都继承它自动拥有时间字段classBase(DeclarativeBase):create_time:Mapped[datetime]mapped_column(DateTime,insert_defaultfunc.now(),comment创建时间)update_time:Mapped[datetime]mapped_column(DateTime,insert_defaultfunc.now(),onupdatefunc.now(),comment修改时间)# 图书模型对应数据库的 book 表classBook(Base):__tablename__book# 数据表名# 字段定义id:Mapped[int]mapped_column(primary_keyTrue,autoincrementTrue,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出版社,# 注释nullableFalse,# 是否允许为空default未知,# 默认值uniqueTrue# 是否唯一)三、核心业务实现main.py1. 项目初始化 自动建表新版 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 实例appFastAPI(title异步图书管理系统,lifespanlifespan)注意启动建表的函数依旧可以当成模板钩子函数 系统在特定时间点自动喊你运行的函数2.12 种核心接口实现按照需求实现查询、新增、修改、删除、分页、统计、模糊查询等功能1查询类接口execute(select(…))# 1. 获取所有图书app.get(/books)asyncdefget_all_books(db:AsyncSessionDepends(get_db)): Depends(get_db)自动获取数据库连接 不用你手动开连接、关连接 全部自动管理 最终得到一个叫 db 的对象用来操作数据库 resultawaitdb.execute(select(Book))booksresult.scalars().all()return{data:books}详细注释result类型AsyncResultSQLAlchemy 2.0 异步结果对象本质数据库查询结果集的封装容器包含原始行数据、元数据、游标状态内部以行元组形式存储。不能直接序列化、不能直接遍历为对象列表2.scalars()提取每行的第一个标量值将 Result → ScalarResult效果(Book对象,) → Book对象去掉外层元组3.await只用于【需要等待 I/O 操作】的代码I/O 和数据库、网络、文件打交道的操作所以以上代码只有db.execute(…)发送 SQL 给 MySQL需要添加awaitexecute(select(…).limit(num))# 2. 获取第一本图书app.get(/books/first)asyncdefget_first_book(db:AsyncSessionDepends(get_db)):resultawaitdb.execute(select(Book).limit(1))bookresult.scalar_one_or_none()ifnotbook:raiseHTTPException(status_code404,detail暂无数据)return{data:book}详细注释scalar_one_or_none()有且仅有一条数据 → 返回 Book 对象无数据 / 多条数据 → 返回 Noneget()# 3. 根据主键 get() 查询单本图书app.get(/books/get/{book_id})asyncdefget_book_by_id(book_id:int,db:AsyncSessionDepends(get_db)):bookawaitdb.get(Book,book_id)ifnotbook:raiseHTTPException(status_code404,detail书籍不存在)return{data:book}详细注释get()根据主键查询的快捷方法# 4. 路径参数查询图书app.get(/books/{book_id})asyncdefget_book_by_path(book_id:int,db:AsyncSessionDepends(get_db)):resultawaitdb.execute(select(Book).where(Book.idbook_id))return{data:result.scalar_one_or_none()}# 5. 条件查询价格≥200app.get(/books/price/ge200)asyncdefget_books_price_ge200(db:AsyncSessionDepends(get_db)):resultawaitdb.execute(select(Book).where(Book.price200))return{data:result.scalars().all()}# 6. 模糊查询作者以周开头app.get(/books/author/zhou)asyncdefget_books_author_zhou(db:AsyncSessionDepends(get_db)):resultawaitdb.execute(select(Book).where(Book.author.like(周%)))return{data:result.scalars().all()}where(条件1, 条件2, 条件3)# 7. 多条件查询作者周开头 价格100app.get(/books/zhou/price/gt100)asyncdefget_books_zhou_and_price(db:AsyncSessionDepends(get_db)):resultawaitdb.execute(select(Book).where(Book.author.like(周%),Book.price100))return{data:result.scalars().all()}# 8. 统计查询总数 平均价格app.get(/books/stat)asyncdefget_books_stat(db:AsyncSessionDepends(get_db)):countawaitdb.scalar(select(func.count(Book.id)))avg_priceawaitdb.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:int1,size:int10,db:AsyncSessionDepends(get_db)):#计算偏移量offset(page-1)*size resultawaitdb.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_code201)asyncdefcreate_book(book_in:BookCreate,db:AsyncSessionDepends(get_db)):new_bookBook(**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:AsyncSessionDepends(get_db)):bookawaitdb.get(Book,book_id)ifnotbook:raiseHTTPException(status_code404)fork,vinbook_in.dict(exclude_unsetTrue).items():setattr(book,k,v)awaitdb.commit()return{msg:修改成功}详细注释setattr(book, k, v)给 book 对象的字段赋值book_in.dict(exclude_unsetTrue)把前端传过来的数据转成字典exclude_unsetTrue只取前端真正传过来的字段没传的字段忽略不改# 12. 删除图书app.delete(/books/{book_id})asyncdefdelete_book(book_id:int,db:AsyncSessionDepends(get_db)):bookawaitdb.get(Book,book_id)ifnotbook:raiseHTTPException(status_code404)awaitdb.delete(book)awaitdb.commit()return{msg:删除成功}运行项目if__name____main__:importuvicorn uvicorn.run(main:app,host127.0.0.1,port8000,reloadTrue)四、项目运行与测试启动项目直接运行 main.py控制台显示服务启动成功即可。可视化接口测试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