从零实战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_defaultfunc.now(),defaultfunc.now(),comment创建时间)update_time:Mapped[datetime]mapped_column(DateTime,insert_defaultfunc.now(),defaultfunc.now(),onupdatefunc.now(),comment更新时间)classBook(Base):__tablename__bookid:Mapped[int]mapped_column(primary_keyTrue,commentID)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__userid:Mapped[int]mapped_column(primary_keyTrue,commentID)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_URLmysqlaiomysql://root:123456localhost:3306/FastAPI_first?charsetutf8mb4async_enginecreate_async_engine(ASYNC_DATABASE_URL,echoTrue,pool_size20,max_overflow10,pool_recycle3600)# 创建会话工厂Async_sessionLocalasync_sessionmaker(bindasync_engine,class_AsyncSession,expire_on_commitFalse# 关键避免提交后对象失效)五、核心步骤三依赖注入 —— 获取数据库会话fromfastapiimportDepends,HTTPExceptionasyncdefget_db():asyncwithAsync_sessionLocal()assession:try:yieldsessionexceptException:awaitsession.rollback()raisefinally:awaitsession.close()从此每个路由函数都可以通过db: AsyncSession Depends(get_db)自动获取数据库连接六、核心步骤四CRUD 完整路由实现精华1. 查询所有书籍GET /book/booksapp.get(/book/books)asyncdefget_book_list(db:AsyncSessionDepends(get_db)):resultawaitdb.execute(select(Book))booksresult.scalars().all()returnbooks2. 查询单个书籍按 IDapp.get(/book/book/{id})asyncdefget_book_by_id(id:int,db:AsyncSessionDepends(get_db)):resultawaitdb.execute(select(Book).where(Book.idid))bookresult.scalars().first()ifnotbook:raiseHTTPException(status_code404,detail书籍不存在)returnbook3. 模糊查询LIKEapp.get(/book/search_book_like)asyncdefsearch_book_like(name:str,db:AsyncSessionDepends(get_db)):resultawaitdb.execute(select(Book).where(Book.name.like(f%{name}%)))booksresult.scalars().all()returnbooks4. 聚合查询MAX、SUM、AVGapp.get(/book/count)asyncdefget_book_count(db:AsyncSessionDepends(get_db)):resultawaitdb.execute(select(func.sum(Book.price)))total_priceresult.scalar()return{total_price:total_price}5. 分页查询OFFSET LIMITapp.get(/book/page)asyncdefget_page(page:int1,page_size:int3,db:AsyncSessionDepends(get_db)):resultawaitdb.execute(select(Book).offset((page-1)*page_size).limit(page_size))booksresult.scalars().all()returnbooks6. 增加书籍POST /book/add_bookfrompydanticimportBaseModelclassBookCreate(BaseModel):name:strauthor:strprice:floatpublisher:strapp.post(/book/add_book)asyncdefadd_book(book:BookCreate,db:AsyncSessionDepends(get_db)):book_objBook(**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:strapp.put(/book/update_book/{id})asyncdefupdate_book(id:int,book:BookUpdate,db:AsyncSessionDepends(get_db)):book_objawaitdb.get(Book,id)ifnotbook_obj:raiseHTTPException(status_code404,detail书籍不存在)book_obj.namebook.name book_obj.authorbook.author book_obj.pricebook.price book_obj.publisherbook.publisherawaitdb.commit()returnbook_obj8. 删除书籍DELETE /book/delete_book/{id}app.delete(/book/delete_book/{id})asyncdefdelete_book(id:int,db:AsyncSessionDepends(get_db)):book_objawaitdb.get(Book,id)ifnotbook_obj:raiseHTTPException(status_code404,detail书籍不存在)awaitdb.delete(book_obj)awaitdb.commit()return{message:删除成功}七、核心亮点总结技能说明✅Depends(get_db)自动注入数据库会话避免重复代码✅select().where()SQL 查询构建器比原生 SQL 更安全✅scalars().all()提取 ORM 对象列表✅await db.commit()提交事务✅Pydantic Model请求数据自动校验✅HTTPException统一错误处理✅expire_on_commitFalse防止“对象失效”问题