MySQL 与 FastAPI 交互教程_fastapi mysql

目录

• • 1. 使用 Docker 启动 MySQL
  • 2. 创建 FastAPI 应用
  • • 安装必要的依赖
    • 创建项目结构
    • 创建数据库连接模块
    • 创建数据模型
    • 创建 Pydantic 模型（用于请求和响应）
    • 创建主应用
  • 3. 运行和测试应用
  • • 启动应用
    • 访问 API 文档
  • 4. 测试 API 端点
  • • 创建用户
    • 获取所有用户
    • 获取特定用户

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_enginefrom sqlalchemy.ext.declarative import declarative_basefrom sqlalchemy.orm import sessionmaker# MySQL 连接 URLSQLALCHEMY_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 db finally: db.close()

创建数据模型

from sqlalchemy import Column, Integer, String, Boolean, DateTimefrom 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 Optionalfrom datetime import datetimefrom pydantic import BaseModel, EmailStr# 用户基础模型class UserBase(BaseModel): username: str email: str is_active: Optional[bool] = True# 创建用户请求模型class UserCreate(UserBase): password: str# 用户响应模型class User(UserBase): id: int created_at: datetime updated_at: Optional[datetime] = None class Config: orm_mode = True

创建主应用

from fastapi import FastAPI, Depends, HTTPExceptionfrom sqlalchemy.orm import Sessionfrom typing import Listimport uvicornfrom . import models, schemasfrom .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.username db_user.email = user.email db_user.hashed_password = user.password # 实际应用中应该哈希处理 db_user.is_active = user.is_active db.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-mysqluvicorn 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\'