【FastAPI】 + SQLAlchemy 异步 ORM 实现完整 CRUD 操作

🚀从零实战：FastAPI + SQLAlchemy 异步 ORM 实现完整 CRUD 操作（附完整代码）

一、为什么要学「FastAPI + SQLAlchemy 异步 ORM」？

在现代 Web 服务中，数据库是核心组件。然而，传统同步操作（如MySQLdb）在高并发场景下容易成为瓶颈。

FastAPI + SQLAlchemy 异步 ORM 的组合，具备以下优势：

• ✅ 高性能（异步非阻塞 I/O）
• ✅ 代码简洁（面向对象建模）
• ✅ 安全可靠（SQL 注入防护）
• ✅ 自动生成 API 文档（Swagger UI）
• ✅ 与 Pydantic 完美结合，实现数据验证

本文将基于你今天学习的完整内容，带你从零实现一个完整的图书管理系统的 CRUD 功能！

二、环境准备（开发前置）

pipinstallfastapi uvicorn sqlalchemy sqlalchemy-asyncio aiomysql pydantic

✅ 推荐使用uvicorn启动服务：

uvicorn main:app--reload

三、核心步骤一：定义模型类（ORM 映射）

fromsqlalchemyimportDateTime,funcfromsqlalchemy.ormimportDeclarativeBase,Mapped,mapped_columnfromdatetimeimportdatetimeclassBase(DeclarativeBase):# 自动生成创建与更新时间create_time:Mapped[datetime]=mapped_column(DateTime,insert_default=func.now(),default=func.now(),comment="创建时间")update_time:Mapped[datetime]=mapped_column(DateTime,insert_default=func.now(),default=func.now(),onupdate=func.now(),comment="更新时间")classBook(Base):__tablename__="book"id:Mapped[int]=mapped_column(primary_key=True,comment="ID")name: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="出版社")classUser(Base):__tablename__="user"id:Mapped[int]=mapped_column(primary_key=True,comment="ID")username:Mapped[str]=mapped_column(String(255),comment="用户名")password:Mapped[str]=mapped_column(String(255),comment="密码")email:Mapped[str]=mapped_column(String(255),comment="邮箱")

✅ ✅ 小技巧：所有模型继承Base，自动拥有时间字段！

四、核心步骤二：创建数据库引擎 & 会话工厂

fromsqlalchemy.ext.asyncioimportcreate_async_engine,async_sessionmaker ASYNC_DATABASE_URL="mysql+aiomysql://root:123456@localhost:3306/FastAPI_first?charset=utf8mb4"async_engine=create_async_engine(ASYNC_DATABASE_URL,echo=True,pool_size=20,max_overflow=10,pool_recycle=3600)# 创建会话工厂Async_sessionLocal=async_sessionmaker(bind=async_engine,class_=AsyncSession,expire_on_commit=False# 关键！避免提交后对象失效！)

五、核心步骤三：依赖注入 —— 获取数据库会话

fromfastapiimportDepends,HTTPExceptionasyncdefget_db():asyncwithAsync_sessionLocal()assession:try:yieldsessionexceptException:awaitsession.rollback()raisefinally:awaitsession.close()

从此，每个路由函数都可以通过db: AsyncSession = Depends(get_db)自动获取数据库连接！

六、核心步骤四：CRUD 完整路由实现（精华！）

1. 查询所有书籍（GET /book/books）

@app.get("/book/books")asyncdefget_book_list(db:AsyncSession=Depends(get_db)):result=awaitdb.execute(select(Book))books=result.scalars().all()returnbooks

2. 查询单个书籍（按 ID）

@app.get("/book/book/{id}")asyncdefget_book_by_id(id:int,db:AsyncSession=Depends(get_db)):result=awaitdb.execute(select(Book).where(Book.id==id))book=result.scalars().first()ifnotbook:raiseHTTPException(status_code=404,detail="书籍不存在")returnbook

3. 模糊查询（LIKE）

@app.get("/book/search_book_like")asyncdefsearch_book_like(name:str,db:AsyncSession=Depends(get_db)):result=awaitdb.execute(select(Book).where(Book.name.like(f"%{name}%")))books=result.scalars().all()returnbooks

4. 聚合查询（MAX、SUM、AVG）

@app.get("/book/count")asyncdefget_book_count(db:AsyncSession=Depends(get_db)):result=awaitdb.execute(select(func.sum(Book.price)))total_price=result.scalar()return{"total_price":total_price}

5. 分页查询（OFFSET + LIMIT）

@app.get("/book/page")asyncdefget_page(page:int=1,page_size:int=3,db:AsyncSession=Depends(get_db)):result=awaitdb.execute(select(Book).offset((page-1)*page_size).limit(page_size))books=result.scalars().all()returnbooks

6. 增加书籍（POST /book/add_book）

frompydanticimportBaseModelclassBookCreate(BaseModel):name:strauthor:strprice:floatpublisher:str@app.post("/book/add_book")asyncdefadd_book(book:BookCreate,db:AsyncSession=Depends(get_db)):book_obj=Book(**book.dict())db.add(book_obj)awaitdb.commit()awaitdb.refresh(book_obj)returnbook_obj

自动校验类型，无须手动判断数据合法性

7. 更新书籍（PUT /book/update_book/{id}）

classBookUpdate(BaseModel):name:strauthor:strprice:floatpublisher:str@app.put("/book/update_book/{id}")asyncdefupdate_book(id:int,book:BookUpdate,db:AsyncSession=Depends(get_db)):book_obj=awaitdb.get(Book,id)ifnotbook_obj:raiseHTTPException(status_code=404,detail="书籍不存在")book_obj.name=book.name book_obj.author=book.author book_obj.price=book.price book_obj.publisher=book.publisherawaitdb.commit()returnbook_obj

8. 删除书籍（DELETE /book/delete_book/{id}）

@app.delete("/book/delete_book/{id}")asyncdefdelete_book(id:int,db:AsyncSession=Depends(get_db)):book_obj=awaitdb.get(Book,id)ifnotbook_obj:raiseHTTPException(status_code=404,detail="书籍不存在")awaitdb.delete(book_obj)awaitdb.commit()return{"message":"删除成功"}

七、核心亮点总结