第 7 章: 数据库

---

本章概述

学习如何将FastAPI连接到关系数据库（如MySQL、PostgreSQL、SQLite）。

---

7.1 安装依赖

pip install sqlalchemy
pip install pymysql  # MySQL驱动
# 或
pip install psycopg2  # PostgreSQL驱动
# 或
pip install aiosqlite  # SQLite异步驱动

---

7.2 配置连接

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

SQLALCHEMY_DATABASE_URL = "mysql+pymysql://user:password@localhost/dbname"

engine = create_engine(SQLALCHEMY_DATABASE_URL)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

Base = declarative_base()

---

7.3 创建模型（数据库表）

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

class 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)
    password = Column(String(100))

---

7.4 创建表

# 在应用启动时创建表
Base.metadata.create_all(bind=engine)

---

7.5 在路由中注入数据库会话

from fastapi import Depends
from sqlalchemy.orm import Session
from database import SessionLocal

def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

@app.get("/users/")
def read_users(db: Session = Depends(get_db)):
    users = db.query(User).all()
    return users

---

7.6 CRUD操作

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

# 读取
def get_user(db: Session, user_id: int):
    return db.query(User).filter(User.id == user_id).first()

# 更新
def update_user(db: Session, user_id: int, user: UserUpdate):
    db_user = db.query(User).filter(User.id == user_id).first()
    for key, value in user.dict(exclude_unset=True).items():
        setattr(db_user, key, value)
    db.commit()
    return db_user

# 删除
def delete_user(db: Session, user_id: int):
    db_user = db.query(User).filter(User.id == user_id).first()
    db.delete(db_user)
    db.commit()

---

7.7 关系

一对多关系

class User(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True)
    tasks = relationship("Task", back_populates="owner")

class Task(Base):
    __tablename__ = "tasks"
    id = Column(Integer, primary_key=True)
    owner_id = Column(Integer, ForeignKey("users.id"))
    owner = relationship("User", back_populates="tasks")

多对多关系

task_tags = Table(
    "task_tags",
    Base.metadata,
    Column("task_id", Integer, ForeignKey("tasks.id")),
    Column("tag_id", Integer, ForeignKey("tags.id"))
)

class Task(Base):
    __tablename__ = "tasks"
    id = Column(Integer, primary_key=True)
    tags = relationship("Tag", secondary=task_tags, back_populates="tasks")

class Tag(Base):
    __tablename__ = "tags"
    id = Column(Integer, primary_key=True)
    tasks = relationship("Task", secondary=task_tags, back_populates="tags")

---

7.8 分页

@app.get("/tasks/")
def read_tasks(skip: int = 0, limit: int = 10, db: Session = Depends(get_db)):
    tasks = db.query(Task).offset(skip).limit(limit).all()
    return tasks

---

小结

本章学习了： - ✅ 安装SQLAlchemy和数据库驱动 - ✅ 配置数据库连接 - ✅ 创建数据库模型 - ✅ 使用依赖注入管理数据库会话 - ✅ 实现CRUD操作 - ✅ 数据库关系（一对多、多对多） - ✅ 分页功能