MySQL 与 FastAPI 交互教程

1. 使用 Docker 启动 MySQL

首先，我们需要使用 Docker 启动一个 MySQL 容器：

docker run --name mysql-server -e MYSQL_ROOT_PASSWORD=password -e MYSQL_DATABASE=fastapi_db -p 3306:3306 -d mysql:8.0

这个命令会：

• 创建一个名为 mysql-server 的容器
• 设置 root 密码为 password
• 创建一个名为 fastapi_db 的数据库
• 将容器的 3306 端口映射到主机的 3306 端口
• 使用 MySQL 8.0 版本的镜像
• 在后台运行容器

2. 创建 FastAPI 应用

安装必要的依赖

uv pip install fastapi pymysql sqlalchemy uvicorn cryptography

创建项目结构

fastapi-mysql/
├── app/
│   ├── __init__.py
│   ├── main.py
│   ├── database.py
│   ├── models.py
│   └── schemas.py

创建数据库连接模块

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker# MySQL 连接 URL
SQLALCHEMY_DATABASE_URL = "mysql+pymysql://root:password@localhost:3306/fastapi_db"# 创建 SQLAlchemy 引擎
engine = create_engine(SQLALCHEMY_DATABASE_URL)# 创建会话类
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)# 创建基类
Base = declarative_base()# 获取数据库会话的依赖项
def get_db():db = SessionLocal()try:yield dbfinally:db.close()

创建数据模型

from sqlalchemy import Column, Integer, String, Boolean, DateTime
from sqlalchemy.sql import funcfrom .database import Baseclass User(Base):__tablename__ = "users"id = Column(Integer, primary_key=True, index=True)username = Column(String(50), unique=True, index=True)email = Column(String(100), unique=True, index=True)hashed_password = Column(String(100))is_active = Column(Boolean, default=True)created_at = Column(DateTime(timezone=True), server_default=func.now())updated_at = Column(DateTime(timezone=True), onupdate=func.now())

创建 Pydantic 模型（用于请求和响应）

from typing import Optional
from datetime import datetime
from pydantic import BaseModel, EmailStr# 用户基础模型
class UserBase(BaseModel):username: stremail: stris_active: Optional[bool] = True# 创建用户请求模型
class UserCreate(UserBase):password: str# 用户响应模型
class User(UserBase):id: intcreated_at: datetimeupdated_at: Optional[datetime] = Noneclass Config:orm_mode = True

创建主应用

from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy.orm import Session
from typing import List
import uvicornfrom . import models, schemas
from .database import engine, get_db# 创建数据库表
models.Base.metadata.create_all(bind=engine)# 创建 FastAPI 应用
app = FastAPI(title="FastAPI MySQL 示例")# 创建用户
@app.post("/users/", response_model=schemas.User)
def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)):# 检查用户名是否已存在db_user = db.query(models.User).filter(models.User.username == user.username).first()if db_user:raise HTTPException(status_code=400, detail="用户名已被注册")# 检查邮箱是否已存在db_user = db.query(models.User).filter(models.User.email == user.email).first()if db_user:raise HTTPException(status_code=400, detail="邮箱已被注册")# 创建新用户（注意：实际应用中应该对密码进行哈希处理）db_user = models.User(username=user.username,email=user.email,hashed_password=user.password,  # 实际应用中应该哈希处理is_active=user.is_active)db.add(db_user)db.commit()db.refresh(db_user)return db_user# 获取所有用户
@app.get("/users/", response_model=List[schemas.User])
def read_users(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):users = db.query(models.User).offset(skip).limit(limit).all()return users# 根据 ID 获取用户
@app.get("/users/{user_id}", response_model=schemas.User)
def read_user(user_id: int, db: Session = Depends(get_db)):db_user = db.query(models.User).filter(models.User.id == user_id).first()if db_user is None:raise HTTPException(status_code=404, detail="用户不存在")return db_user# 更新用户
@app.put("/users/{user_id}", response_model=schemas.User)
def update_user(user_id: int, user: schemas.UserCreate, db: Session = Depends(get_db)):db_user = db.query(models.User).filter(models.User.id == user_id).first()if db_user is None:raise HTTPException(status_code=404, detail="用户不存在")# 更新用户信息db_user.username = user.usernamedb_user.email = user.emaildb_user.hashed_password = user.password  # 实际应用中应该哈希处理db_user.is_active = user.is_activedb.commit()db.refresh(db_user)return db_user# 删除用户
@app.delete("/users/{user_id}", response_model=schemas.User)
def delete_user(user_id: int, db: Session = Depends(get_db)):db_user = db.query(models.User).filter(models.User.id == user_id).first()if db_user is None:raise HTTPException(status_code=404, detail="用户不存在")db.delete(db_user)db.commit()return db_userif __name__ == "__main__":uvicorn.run("app.main:app", host="0.0.0.0", port=8000, reload=True)

3. 运行和测试应用

启动应用

cd fastapi-mysql
uvicorn app.main:app --reload

应用将在 http://localhost:8000 上运行。

访问 API 文档

FastAPI 自动生成的 API 文档可以在以下地址访问：

• Swagger UI: http://localhost:8000/docs
• ReDoc: http://localhost:8000/redoc

4. 测试 API 端点

你可以使用 Swagger UI 或者 curl 命令来测试 API：

创建用户

curl -X 'POST' \'http://localhost:8000/users/' \-H 'accept: application/json' \-H 'Content-Type: application/json' \-d '{"username": "testuser","email": "test@example.com","is_active": true,"password": "password123"
}'

获取所有用户

curl -X 'GET' \'http://localhost:8000/users/' \-H 'accept: application/json'

获取特定用户

curl -X 'GET' \'http://localhost:8000/users/1' \-H 'accept: application/json'