大家好我是Java1234_小锋老师最近更新《2027版 一天学会 FastAPI Python web开发 视频教程(无废话版)》专辑感谢大家支持。本课程主要介绍和讲解FastAPI简介HelloWorld实现自动生成交互式API文档路径参数查询参数请求体参数校验响应模型 表单数据和模型中间件 依赖注入集成SQLAlchemy ORM 操作数据库集成Pydantic数据校验等视频教程课件源码打包下载链接https://pan.baidu.com/s/1_NzaNr0Wln6kv1rdiQnUTg提取码0000集成SQLAlchemy ORM 操作数据库FastAPI 是目前非常流行的 Python 异步 Web 框架而 SQLAlchemy 则是 Python 生态中最强大、最成熟的 ORM对象关系映射工具。将两者结合可以快速构建出高性能、类型安全且易于维护的 Web 应用。本文将从零开始带你完成 FastAPI 与 SQLAlchemy 的集成实现数据库的建表及完整的增删改查CRUD功能。整体架构fastapi_pro/ ├── main.py # FastAPI 入口已有 ├── database.py # 数据库连接与 Session 管理 ├── models/ │ └── item.py # SQLAlchemy ORM 模型对应表 t_item ├── schemas/ │ └── item.py # Pydantic 模型请求/响应校验 ├── crud/ │ └── item.py # 数据库 CRUD 操作 └── routers/ └── item.py # Item 相关 API 路由数据流向如下HTTP 请求 → FastAPI 路由 → Pydantic 校验 → CRUD 层 → SQLAlchemy Session → MySQL这与项目中已有的写法一致路由层负责接收参数Pydantic 负责校验HTTPException负责错误响应response_model负责过滤返回字段。一、安装依赖在项目虚拟环境中安装pip install sqlalchemy pymysql cryptography包名作用sqlalchemyORM 框架pymysqlMySQL 驱动cryptographypymysql 连接加密时可能需要二、创建数据库在 MySQL 中先创建数据库库名以db_开头CREATE DATABASE IF NOT EXISTS db_fastapi_pro DEFAULT CHARACTER SET utf8mb4 DEFAULT COLLATE utf8mb4_unicode_ci;三、数据库连接配置新建database.py统一管理引擎、Session 与依赖注入from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker, DeclarativeBase ​ # MySQL 连接串密码按项目约定为 123456 SQLALCHEMY_DATABASE_URL ( mysqlpymysql://root:123456127.0.0.1:3308/db_fastapi_pro?charsetutf8mb4 ) ​ # 创建引擎 engine create_engine( SQLALCHEMY_DATABASE_URL, pool_pre_pingTrue, # 连接池自动检测断线重连 pool_recycle3600, # 每小时回收连接避免 MySQL 8 小时超时 ) ​ # Session 工厂 SessionLocal sessionmaker(autocommitFalse, autoflushFalse, bindengine) ​ ​ # SQLAlchemy 2.0 声明式基类 class Base(DeclarativeBase): pass ​ ​ def get_db(): FastAPI 依赖每个请求独立 Session用完自动关闭 db SessionLocal() try: yield db # yield作用 暂停在这里把 db 交给 FastAPI; finally: db.close()要点说明get_db()通过yield实现依赖注入与main3.py中Depends(common_parameters)的用法相同。每个 HTTP 请求使用独立 Session避免线程/协程间共享连接。yield 作用是把 get_db() 变成一个生成器函数FastAPI 的 Depends(get_db) 会按「先拿到资源 → 执行接口 → 再清理资源」这个顺序运行。四、定义 ORM 模型建表新建models/item.py定义与t_item表映射的 ORM 类from datetime import datetime ​ from sqlalchemy import String, Float, DateTime, Integer, Text from sqlalchemy.orm import Mapped, mapped_column ​ from database import Base ​ ​ class ItemModel(Base): 物品表 ORM 模型对应数据库表 t_item ​ __tablename__ t_item ​ id: Mapped[int] mapped_column(Integer, primary_keyTrue, autoincrementTrue, comment主键ID) name: Mapped[str] mapped_column(String(50), nullableFalse, comment物品名称) price: Mapped[float] mapped_column(Float, nullableFalse, comment价格) description: Mapped[str | None] mapped_column(Text, nullableTrue, comment描述) created_at: Mapped[datetime] mapped_column( DateTime, defaultdatetime.now, comment创建时间 ) updated_at: Mapped[datetime] mapped_column( DateTime, defaultdatetime.now, onupdatedatetime.now, comment更新时间 )自动建表在应用启动时执行Base.metadata.create_all()SQLAlchemy 会根据模型自动创建t_item表from database import engine, Base from models.item import ItemModel # 必须导入否则模型不会注册 ​ # 创建所有表已存在的表不会重复创建 Base.metadata.create_all(bindengine)自动建表刷新sqlyog等价于执行以下 SQLCREATE TABLE t_item ( id INT AUTO_INCREMENT PRIMARY KEY COMMENT 主键ID, name VARCHAR(50) NOT NULL COMMENT 物品名称, price FLOAT NOT NULL COMMENT 价格, description TEXT NULL COMMENT 描述, created_at DATETIME NOT NULL COMMENT 创建时间, updated_at DATETIME NOT NULL COMMENT 更新时间 ) ENGINEInnoDB DEFAULT CHARSETutf8mb4;五、Pydantic 模型请求/响应新建schemas/item.py。字段校验规则参考main.py中已有的ItemCreatefrom datetime import datetime ​ from pydantic import BaseModel, Field, ConfigDict ​ ​ class ItemCreate(BaseModel): 创建物品时的请求体 name: str Field(..., min_length2, max_length50, description物品名称2~50个字符) price: float Field(..., gt0, le9999.99, description价格必须大于0) description: str | None Field(None, max_length200, description可选描述) ​ ​ class ItemUpdate(BaseModel): 更新物品时的请求体字段均可选 name: str | None Field(None, min_length2, max_length50) price: float | None Field(None, gt0, le9999.99) description: str | None Field(None, max_length200) ​ ​ class ItemOut(BaseModel): 返回给客户端的数据结构 id: int name: str price: float description: str | None None created_at: datetime updated_at: datetime ​ model_config ConfigDict(from_attributesTrue)from_attributesTrue允许 Pydantic 直接从 SQLAlchemy ORM 对象读取属性无需手动转字典。六、CRUD 操作新建crud/item.py封装对t_item表的增删改查from sqlalchemy.orm import Session ​ from models.item import ItemModel from schemas.item import ItemCreate, ItemUpdate ​ ​ # -------- 增Create-------- def create_item(db: Session, item: ItemCreate) - ItemModel: db_item ItemModel( nameitem.name, priceitem.price, descriptionitem.description, ) db.add(db_item) db.commit() db.refresh(db_item) return db_item ​ ​ # -------- 查Read-------- def get_item(db: Session, item_id: int) - ItemModel | None: return db.query(ItemModel).filter(ItemModel.id item_id).first() ​ ​ def get_items(db: Session, skip: int 0, limit: int 10) - list[ItemModel]: return db.query(ItemModel).offset(skip).limit(limit).all() ​ ​ # -------- 改Update-------- def update_item(db: Session, item_id: int, item: ItemUpdate) - ItemModel | None: db_item get_item(db, item_id) if not db_item: return None ​ update_data item.model_dump( exclude_unsetTrue) # 排除未设置的字段 model_dump() 会把模型转成普通字典 exclude_unsetTrue 是关键只包含客户端在 JSON 里实际传了的字段没传的字段不会出现在字典里。 for field, value in update_data.items(): setattr(db_item, field, value) # 设置字段值 ​ db.commit() db.refresh(db_item) return db_item ​ ​ # -------- 删Delete-------- def delete_item(db: Session, item_id: int) - bool: db_item get_item(db, item_id) if not db_item: return False ​ db.delete(db_item) db.commit() return True​CRUD 对应 SQL 说明操作ORM 方法等价 SQL增db.add()db.commit()INSERT INTO t_item ...查单条db.query().filter().first()SELECT * FROM t_item WHERE id ? LIMIT 1查列表db.query().offset().limit().all()SELECT * FROM t_item LIMIT ? OFFSET ?改setattr()db.commit()UPDATE t_item SET ... WHERE id ?删db.delete()db.commit()DELETE FROM t_item WHERE id ?七、FastAPI 路由集成新建routers/item.py将 CRUD 暴露为 REST APIfrom typing import Annotated ​ from fastapi import APIRouter, Depends, HTTPException, Query from sqlalchemy.orm import Session ​ from crud import item as item_crud from database import get_db from schemas.item import ItemCreate, ItemUpdate, ItemOut ​ router APIRouter(prefix/items, tags[物品管理]) ​ DbSession Annotated[Session, Depends(get_db)] ​ ​ router.post(/, response_modelItemOut, summary新增物品) def create_item(item: ItemCreate, db: DbSession): return item_crud.create_item(db, item) ​ ​ router.get(/, response_modellist[ItemOut], summary物品列表分页) def list_items( db: DbSession, skip: int Query(0, ge0, title跳过条数), limit: int Query(10, ge1, le100, title返回条数), ): return item_crud.get_items(db, skipskip, limitlimit) ​ ​ router.get(/{item_id}, response_modelItemOut, summary查询单个物品) def read_item(item_id: int, db: DbSession): db_item item_crud.get_item(db, item_id) if not db_item: raise HTTPException(status_code404, detailItem项不存在) return db_item ​ ​ router.put(/{item_id}, response_modelItemOut, summary更新物品) def update_item(item_id: int, item: ItemUpdate, db: DbSession): db_item item_crud.update_item(db, item_id, item) if not db_item: raise HTTPException(status_code404, detailItem项不存在) return db_item ​ ​ router.delete(/{item_id}, summary删除物品) def remove_item(item_id: int, db: DbSession): success item_crud.delete_item(db, item_id) if not success: raise HTTPException(status_code404, detailItem项不存在) return {message: 删除成功, item_id: item_id}404 错误处理方式与main.py中read_item接口一致# main.py 中的写法 if item_id ! 1: raise HTTPException(status_code404, detailItem项不存在)八、注册路由并启动在main.py或新建main_db.py中整合from contextlib import asynccontextmanager ​ from fastapi import FastAPI ​ from database import engine, Base from models.item import ItemModel from routers.item import router as item_router ​ ​ # 这段代码是 FastAPI 的应用生命周期lifespan管理用来在服务启动时做初始化在服务关闭时做清理。 asynccontextmanager # 异步上下文管理器 async def lifespan(app: FastAPI): # 启动时执行 Base.metadata.create_all(bindengine) yield # 应用运行中... # 关闭时执行CtrlC、重启、进程退出 engine.dispose() # 释放数据库连接池 print(应用已关闭) ​ ​ app FastAPI(titleFastAPI SQLAlchemy 示例, lifespanlifespan) ​ # 注册路由 app.include_router(item_router) ​ ​ app.get(/) async def root(): return {message: 你好FastAPI SQLAlchemy!}启动服务uvicorn main_db:app --reload访问 Swagger 文档http://127.0.0.1:8000/docs九、API 测试示例新增测试查询测试单个查询更新操作删除测试