# fastapi-mysql

**Repository Path**: jiayouyc/fastapi-mysql

## Basic Information

- **Project Name**: fastapi-mysql
- **Description**: No description available
- **Primary Language**: Unknown
- **License**: Not specified
- **Default Branch**: main
- **Homepage**: None
- **GVP Project**: No

## Statistics

- **Stars**: 0
- **Forks**: 0
- **Created**: 2026-07-05
- **Last Updated**: 2026-07-05

## Categories & Tags

**Categories**: Uncategorized

**Tags**: None

## README

```
backend/
├── .env                # 数据库地址
├── config.py           # 读取环境变量
├── database.py         # 数据库连接、会话依赖
├── models/
│   └── video.py        # MySQL表映射模型
├── schemas/
│   └── video.py        # Pydantic 校验/返回模型
├── crud/
│   └── video.py        # 纯数据库增删改查逻辑
├── api/
│   └── video.py        # 接口路由
└── main.py             # 入口
```


# FastAPI + SQLAlchemy 标准项目 CRUD 完整示例（企业通用分层）
技术栈：异步SQLAlchemy + MySQL + Pydantic，和你抖音爬虫项目结构完全一致
目录结构
```
backend/
├── .env                # 数据库地址
├── config.py           # 读取环境变量
├── database.py         # 数据库连接、会话依赖
├── models/
│   └── video.py        # MySQL表映射模型
├── schemas/
│   └── video.py        # Pydantic 校验/返回模型
├── crud/
│   └── video.py        # 纯数据库增删改查逻辑
├── api/
│   └── video.py        # 接口路由
└── main.py             # 入口
```

## 1. .env
```env
MYSQL_DATABASE_URI=mysql+asyncmy://root:123456@127.0.0.1:3306/douyin_scraper?charset=utf8mb4
```

## 2. config.py
```python
from pydantic_settings import BaseSettings

class Settings(BaseSettings):
    MYSQL_DATABASE_URI: str

    class Config:
        env_file = ".env"

settings = Settings()
```

## 3. database.py 数据库底层
```python
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import declarative_base, sessionmaker
from config import settings

engine = create_async_engine(settings.MYSQL_DATABASE_URI, echo=False)
AsyncSessionLocal = sessionmaker(bind=engine, class_=AsyncSession, expire_on_commit=False)
Base = declarative_base()

# 接口依赖，自动获取数据库会话
async def get_db():
    async with AsyncSessionLocal() as db:
        yield db
```

## 4. models/video.py 数据库表映射（真实mysql表）
```python
from sqlalchemy import Column, Integer, String, DateTime
from database import Base

class Video(Base):
    __tablename__ = "videos"
    id = Column(Integer, primary_key=True, autoincrement=True, comment="主键")
    aweme_id = Column(String(100), unique=True, nullable=False, comment="作品ID")
    author_name = Column(String(100), comment="作者名")
    author_sec_uid = Column(String(100), comment="作者UID")
    description = Column(String(1000), comment="视频文案")
    crawl_time = Column(DateTime, comment="爬取时间")
```

## 5. schemas/video.py Pydantic 数据模型
```python
from pydantic import BaseModel
from typing import Optional
from datetime import datetime

# 新增视频入参
class VideoCreate(BaseModel):
    aweme_id: str
    author_name: Optional[str] = None
    author_sec_uid: Optional[str] = None
    description: Optional[str] = None

# 修改视频入参
class VideoUpdate(BaseModel):
    author_name: Optional[str] = None
    author_sec_uid: Optional[str] = None
    description: Optional[str] = None

# 返回给前端模型
class VideoResp(BaseModel):
    id: int
    aweme_id: str
    author_name: Optional[str]
    author_sec_uid: Optional[str]
    description: Optional[str]
    crawl_time: Optional[datetime]

    class Config:
        from_attributes = True
```

## 6. crud/video.py 【核心增删改查封装】
只写数据库操作，不处理接口、异常
```python
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select, update, delete
from models.video import Video
from schemas.video import VideoCreate, VideoUpdate

# 1. 新增 CREATE
async def create_video(db: AsyncSession, data: VideoCreate):
    obj = Video(**data.model_dump())
    db.add(obj)
    await db.commit()
    await db.refresh(obj)
    return obj

# 2. 根据ID单条查询 READ
async def get_video_by_id(db: AsyncSession, vid: int):
    stmt = select(Video).where(Video.id == vid)
    res = await db.execute(stmt)
    return res.scalar_one_or_none()

# 3. 查询全部（分页简化版）
async def list_video(db: AsyncSession, page: int = 1, size: int = 20):
    offset = (page - 1) * size
    stmt = select(Video).offset(offset).limit(size)
    res = await db.execute(stmt)
    return res.scalars().all()

# 4. 更新 UPDATE
async def update_video(db: AsyncSession, vid: int, update_info: VideoUpdate):
    # exclude_unset：只更新前端传了的字段
    update_dict = update_info.model_dump(exclude_unset=True)
    stmt = update(Video).where(Video.id == vid).values(**update_dict)
    await db.execute(stmt)
    await db.commit()
    return await get_video_by_id(db, vid)

# 5. 删除 DELETE
async def delete_video(db: AsyncSession, vid: int):
    stmt = delete(Video).where(Video.id == vid)
    await db.execute(stmt)
    await db.commit()
    return True
```

## 7. api/video.py 接口层（接收前端请求）
```python
from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.ext.asyncio import AsyncSession
from database import get_db
from crud.video import create_video, get_video_by_id, list_video, update_video, delete_video
from schemas.video import VideoCreate, VideoUpdate, VideoResp

router = APIRouter(prefix="/video", tags=["视频管理"])

# 新增
@router.post("/add", response_model=VideoResp)
async def add_video(form: VideoCreate, db: AsyncSession = Depends(get_db)):
    return await create_video(db, form)

# 根据id查询
@router.get("/{vid}", response_model=VideoResp)
async def get_single(vid: int, db: AsyncSession = Depends(get_db)):
    video = await get_video_by_id(db, vid)
    if not video:
        raise HTTPException(status_code=404, detail="视频不存在")
    return video

# 分页列表
@router.get("/list", response_model=list[VideoResp])
async def get_list(page: int = 1, size: int = 20, db: AsyncSession = Depends(get_db)):
    return await list_video(db, page, size)

# 修改
@router.put("/{vid}", response_model=VideoResp)
async def edit_video(vid: int, form: VideoUpdate, db: AsyncSession = Depends(get_db)):
    video = await get_video_by_id(db, vid)
    if not video:
        raise HTTPException(status_code=404, detail="视频不存在")
    return await update_video(db, vid, form)

# 删除
@router.delete("/{vid}")
async def del_video(vid: int, db: AsyncSession = Depends(get_db)):
    video = await get_video_by_id(db, vid)
    if not video:
        raise HTTPException(status_code=404, detail="视频不存在")
    await delete_video(db, vid)
    return {"msg": "删除成功"}
```

## 8. main.py 项目入口，注册路由
```python
from fastapi import FastAPI
from api.video import router as video_router

app = FastAPI(title="MySQL CRUD Demo")

# 注册视频接口
app.include_router(video_router)
```

# 运行测试
启动命令
```bash
uvicorn main:app --reload
```
访问接口文档：`http://127.0.0.1:8000/docs`
可以可视化测试新增、查询、修改、删除。

# 标准项目分层逻辑总结
1. 前端请求 → main.py 路由分发
2. api层：接收参数、判空、抛404，调用CRUD
3. crud层：只写SQLAlchemy数据库操作，无业务判断
4. models：映射真实MySQL数据表
5. schemas：参数校验、控制返回字段
6. database/config：统一管理数据库连接，全局复用






## 需要安装的全部依赖包
```bash
pip install fastapi uvicorn sqlalchemy asyncmy pydantic python-dotenv
```

### 每个包作用拆分
1. **fastapi**：Web接口框架主体
2. **uvicorn**：ASGI异步服务启动器，用来跑项目
3. **sqlalchemy**：ORM数据库工具，操作MySQL不用手写SQL
4. **asyncmy**：MySQL异步驱动，配合 `mysql+asyncmy://` 连接串使用（异步高性能）
5. **pydantic**：数据校验模型、读取.env配置
6. **python-dotenv**：加载项目根目录 `.env` 环境变量文件

## 如果你用Conda虚拟环境（你的fastapi环境）
激活环境后再执行安装：
```bash
conda activate fastapi
pip install fastapi uvicorn sqlalchemy asyncmy pydantic python-dotenv
```

## 补充说明
1. 不要用 `pymysql`，示例代码全是**异步写法**，`pymysql` 是同步驱动，混用会报错；
2. 若后续需要迁移数据库（新增字段自动同步到MySQL），额外安装：
```bash
pip install alembic
```
3. 之前你遇到gevent编译报错，如果项目需要celery/gevent，安装稳定版：
```bash
pip install gevent==23.9.1
```