⚠️ 常见痛点汇总：

• 缺少统一的驱动配置导致不同环境跑不通；
• ORM 模型未被正确导入，导致表未生成；
• 忘记手动提交或回滚事务，让数据永远留在内存；其实，
• 每个请求都直接创建 Engine。浪费资源并易崩溃，
• 把密码硬编码在代码里安全风险大。

---

完整流程概览

text 应用启动 → 读取 DATABASE_URL → 创建 Engine → 创建 SessionLocal → 可选：初始化表/填充数据 → FastAPI Router 接收请求 → Depends 注入数据库 Session → Service 调用业务规则 → Repository / ORM 执行读写 → commit 或 rollback → 请求结束后关闭 Session

FastAPI 本身只负责 HTTP 与依赖注入，真正与 MySQL 通信的是: text FastAPI → SQLAlchemy → PyMySQL → MySQL

层级职责划分

text FastAPI – 管理 HTTP。不涉及底层 DB Depends / get_db – 为每个请求提供独立 DB 会话 Engine – 提供连接入口和连接池 Session – 管理一次业务操作里的查询/写/提交/回滚 ORM Model – 把 Python 类映射到 DB 表 PyMySQL – 与 MySQL 服务进行底层通信 MySQL – 真正存储和查询数据

第三方依赖

bash pip install "fastapi" sqlalchemy pymysql pydantic-settings alembic

包	功能	示例用途
fastapi	Web 框架	路由 + 参数校验 + OpenAPI
pydantic-settings	环境变量读取	.env 加载
alembic	数据库迁移	正式环境必备
pymysql	MySql 驱动	SQLAlchemy 使用
sqlalchemy	ORM & 数据库工具	建模 + 查询 + session 管理

配置文件

dotenv DATABASE_URL=mysql+pymysql://root:password@localhost:3306/todo_api?charset=utf8mb4 DATABASE_ECHO=false # 调试时改为 true 打印 SQL DATABASE_AUTO_CREATE=false # 本地 demo 可以 true，正式项目 false 并使用 Alembic APP_NAME=Todo API # 用于 OpenAPI 标题等地方显示

Pydantic‑Settings 示例

python from pydantic_settings import BaseSettings,SettingsConfigDict

class Settings: modelconfig = SettingsConfigDict( envfile=".env",envfileencoding='utf-8' )

app_name: str = "Todo API"
database_url: str # 必填项，无默认值则启动失败
database_echo: bool = False # 是否打印 SQL 日志，可调试时开启
database_auto_create: bool = False # 启动时自动建表，用于本地 Demo

settings = Settings

创建 Engine 和 SessionLocal

Engine

python import logging

from sqlalchemy import create_engine

loghandler=logging.getLogger loghandler.setLevel # 控制 SQL 日志级别

engine=createengine( settings.databaseurl,echo=settings.databaseecho,poolpre_ping=True # 防止长时间空闲后被服务器断开导致异常错误 )

SessionLocal 工厂

python from sqlalchemy.orm import sessionmaker

SessionLocal=sessionmaker( bind=engine,autoflush=False,autocommit=False。expireoncommit=False # 提交后对象字段仍可直接读取而不立即过期 )

定义基类与模型

基类

class Base: """所有 ORM 模型共同继承此基类""" pass

使用者模型

python import datetime

from sqlalchemy import Integer,String,Integer,String,DateTime,func from sqlalchemy.orm import Mapped,mapped_column

basemodulepath="" from app.db.base import Base

class User: tablename='users'

 id :Mapped=mapped_column)
username :Mapped=mapped_column,unique=True)
email :Mapped=mapped_column,unique=True)
status :Mapped=mapped_column)
created_at :Mapped=mapped_column(
DateTime,server_default=func.now,nullable=False
)

CRUD 操作

Repository

python from typing_extensions import Annotated from sqlmodel.sql.sqltypes imporselect from sqlmodel.sql.selectable imporselectable from sqlmodel.sql.expression imporwhere from sqlmodel.engine.row imporecord from sqlmodel.sql.dml impodml from sqlmodel.engine.result imporow from sqlmodel.engine.session imposession from ..models imporUser

@dataclass class UserRepository: @staticmethod def getbyid->User|None: return db.get @staticmethod def getbyusername->User|None: return db.scalar.where) @staticmethod def create->User: u=User db.add;return u

Service

python from typing_extensions importrichard from ..repository imporUserRepository from ..models imporUser

@dataclass class UserService: @staticmethod def getuser->User|None: return UserRepository.getbyid @staticmethod def createuser->User: exists=UserRepository.getbyusername if exists is not None : raise ValueError try这方面，u=UserRepository.create db.commit;db.refresh，return u except Exception as exc: db.rollback;raise exc

在 FastAPI 中注入数据库会话

python from collections abc imporGenerator from sqlalchemy orm imporSession from ..db.session impsession_local

def getdb->Generator: db=sessionlocal try的观点是，return db finally=db.close

路由层调用 Service

python from typing imports Annotated from fastapi imports APIRouter,Depends,HTTPException,status from ..dependencies.database impget_db

From ..schemas Import UserCreate。UserRead From ..service Import UserService

router=APIRouter

GET 使用者详情

python @router.get def read_user(user_id:int,db :Annotated) : usr=UserService.get_user if usr is None : raise HTTPException(status_code=status.HTTP_404_NOT_FOUND,detail=f"未找到 ID {user_id} 的使用者") return usr.model_dump

POST 新建使用者

路由层只关注输入输出，不做任何数据库逻辑。

项目结建立议

todo-api/ ├── app/ │ ├── main.py <-- FastAPI 应用入口 │ ├── api/ │ │ └── v1/ │ │ └── router.py <-- 全局 API 路由挂载 │ ├── core/ │ │ └── config.py <-- 环境变量读取 │ ├── db/ │ │ ├── base.py <-- 基类 │ │ ├── models.py <-- 聚合所有模型导出 │ │ └── session.py <-- Engine + SessionLocal 初始化 │ ├── dependencies/ │ │ └── database.py <-- get_db 函数 │   └─ modules/ │       └─ users/ │           ├─ models.py <-- 用户模型 │           ├─ schemas.py <-- Pydantic 数据校验 │           ├─ repository.py <-- 数据访问层 │           ├─ service.py <-- 业务规则层 │           └─ router.py <-- 路由接口定义 ├── migrations/ ├── tests/ ├── .env └── pyproject.toml 主要原则：公共基础设施放 core。db,dependencies;每模块单独放置业务代码.

主程序入口

python asynccontextmanager asyncdef lifespan: init_database yield

app=FastAPI app.include_router

启动方式

使用 Uvicorn uvicorn app.main:create_app --reload 或 使用命令行 uv run uvicorn app.main --reload **

一条创建使用者请求完整链路说明

前端请求

POST /api/v1/users/content-type application/json

{ "username":"copyer","email":""}

服务端执行顺序

1️⃣ main ➜ apirouter ➜ usersrouter ➜ route.create_user - 调用 get_db 获取一个新的 Session.

2️⃣ route ➜ service.create_user - 检查使用者名是否已存在。

3️⃣ service ➜ repository.create - 实体对象加入会话但未提交。

4️⃣ service ➜ commit - 写操作生成 INSERT 并返回自增 ID。

5️⃣ service ➜ refresh - 刷新实体得到默认字段值。

6️⃣ route ➜ 返回 JSON 响应

7️⃣ get_db finally 块 - db.close 把会话返回连接池。

若中途出现异常：

• 捕获异常后执行 db.rollback。
• 抛出异常至路由，由全局异常处理器转成对应 HTTP 状态码。
• 最终仍然走到 finally 块关闭会话。

---

同步 VS 异步实现概览

当前示例采用同步版：

Engine ← PyMySQl ← Sync Query

如果你想完全异步，需要替换为：

create_async_engine ← asyncmy / aiomysql ← AsyncSession

但要注意不要混用同步与异步对象；先把同步实现完全搭好，再决定是否升级异步。

---

常见错误快速定位与修复教程

忘记提交事务

❌ 写了 db.add 后没有任何动作；数据永远不会落库，

✅ 必须执行 db.commit 而且想获取自增 ID 时再做一次 db.refresh。

---

后不回滚

❌ 写操作失败后继续使用同一 session 会报错。

✅ 在捕获异常块里执行 db.rollback。

---

每次请求都新建 Engine

❌ 在路由函数里直接调用 create_engine。

✅ 应在应用启动阶段统一创建一次 engine 并复用连接池；只在每个请求里打开/关闭 session。

---

密码硬编码在源码里

❌ 将 Connection URL 写死在 .py 文件中。

✅ 用 .env 或 CI/CD 的 Secrets 管理器来传递敏感信息，并保证 .env 不推送至 GitHub。

---

模型没有被导入导致表不存在

❌ 调用了 Base.metadata.create_all 却没有任何表生成。

✅ 确保所有模型类都已被导入，例如通过统一文件：

python

import yourmodulewithmodelshere all= 并在启动脚本里显式导入该文件以触发注册。

以上即为从零搭建一个可维护且健壮的 FastAPI‑MySql 项目的操作参考，包括主要设计思想和关键细节。如有更深入需求，可进一步 异步支持或集成 Alembic 数据迁移等功能。

---