fastapi+mysql 开发web应用

fastapi官网：https://fastapi.tiangolo.com/

本地开发阶段

环境准备

# 创建项目目录
mkdir fastapi-crud && cd fastapi-crud
python -m venv venv
source venv/bin/activate  # Linux/Mac
venv\Scripts\activate     # Windows

# 安装依赖
pip install fastapi uvicorn sqlalchemy pymysql pydantic

项目结构

image.png

代码实现

• 数据库连接 databse.py

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import  declarative_base
from sqlalchemy.orm import  sessionmaker

# 替换成实际的mysql信息
SQLALCHEMY_DATABASE_URL = "mysql+pymysql://root:123456@127.0.0.1:3306/test"
engine = create_engine(SQLALCHEMY_DATABASE_URL)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()

• 数据模型 models.py

from sqlalchemy import Column, Integer, String, DateTime
from .database import Base

class User(Base):
    __tablename__ = "user"
    id = Column(Integer, primary_key=True, index=True)
    name = Column(String(50))
    created_at = Column(DateTime)

• Pydantic模型 schemas.py

from pydantic import BaseModel

class UserCreate(BaseModel):
    name: str

class UserUpdate(BaseModel):
    name: str | None = None

class UserResponse(UserCreate):
    id: int
    class Config:
        orm_mode = True

• CRUD 操作 crud.py

from sqlalchemy.orm import Session
from . import models, schemas


# 创建
def create_user(db: Session, user: schemas.UserCreate):
    db_user = models.User(**user.model_dump())
    db.add(db_user)
    db.commit()
    db.refresh(db_user)
    return db_user


# 查询全部
def get_users(db: Session, skip: int = 0, limit: int = 0):
    return db.query(models.User).offset(skip).limit(limit).all()


# 查询单个
def get_user(db: Session, user_id: int):
    return db.query(models.User).filter(models.User.id == user_id).first()


# 修改
def update_user(db: Session, user_id: int, user: schemas.UserUpdate):
    db_user = get_user(db, user_id)
    if not db_user:
        return None
    for key, value in user.dict(exclude_unset=True).items():
        setattr(db_user, key, value)
    db.commit()
    db.refresh(db_user)
    return db_user


# 删除
def delete_user(db: Session, user_id: int):
    db_user = get_user(db, user_id)
    if not db_user:
        return False
    db.delete(db_user)
    db.commit()
    return True

• FastAPI 主入口 main.py

from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy.orm import Session
from . import models, schemas, crud, database
from typing import List

app = FastAPI()


# 依赖注入数据库会话
def get_db():
    db = database.SessionLocal()
    try:
        yield db
    finally:
        db.close()


# 新增
@app.post("/users/add", response_model=schemas.UserResponse)
def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)):
    return crud.create_user(db, user)


# 查询全部
@app.get("/users", response_model=List[schemas.UserResponse])
def read_users(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    return crud.get_users(db, skip, limit)


# 查询单个
@app.get("/users/{user_id}", response_model=schemas.UserResponse)
def read_user(user_id: int, db: Session = Depends(get_db)):
    user = crud.get_user(db, user_id)
    if not user:
        raise HTTPException(status_code=404, detail="user not found")
    return user


# 修改
@app.post("/users/update/{user_id}")
def update_user(user_id: int, user: schemas.UserUpdate, db: Session = Depends(get_db)):
    user = crud.update_user(db, user_id, user)
    if not user:
        raise HTTPException(status_code=404, detail="user not found")
    return user


# 修改
@app.post("/users/delete/{user_id}")
def update_user(user_id: int, db: Session = Depends(get_db)):
    success = crud.delete_user(db, user_id)
    if not success:
        raise HTTPException(status_code=404, detail="user not found")
    return {"message": "user deleted"}

本地运行测试

访问 http://localhost:8000/docs 测试所有接口

uvicorn app.main:app --reload