SQLAlchemy 集成实验

实验目标

搭建完整的 FastAPI + SQLAlchemy 应用，实现 CRUD 操作和关系查询。

环境准备

pip install fastapi uvicorn sqlalchemy psycopg2-binary alembic

实验 1：数据库配置

创建 database.py：

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

DATABASE_URL = "sqlite:///./test.db"  # 开发环境使用 SQLite

engine = create_engine(
    DATABASE_URL,
    connect_args={"check_same_thread": False}  # SQLite 需要
)

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

Base = declarative_base()

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

实验 2：定义模型

创建 models.py：

from sqlalchemy import Column, Integer, String, DateTime, ForeignKey, Text
from sqlalchemy.orm import relationship
from datetime import datetime
from database import Base

class User(Base):
    __tablename__ = "users"
    
    id = Column(Integer, primary_key=True, index=True)
    email = Column(String, unique=True, index=True, nullable=False)
    username = Column(String, unique=True, index=True, nullable=False)
    hashed_password = Column(String, nullable=False)
    created_at = Column(DateTime, default=datetime.utcnow)
    
    posts = relationship("Post", back_populates="author", cascade="all, delete-orphan")

class Post(Base):
    __tablename__ = "posts"
    
    id = Column(Integer, primary_key=True, index=True)
    title = Column(String, nullable=False)
    content = Column(Text, nullable=False)
    author_id = Column(Integer, ForeignKey("users.id"))
    created_at = Column(DateTime, default=datetime.utcnow)
    
    author = relationship("User", back_populates="posts")

实验 3：Pydantic Schemas

创建 schemas.py：

from pydantic import BaseModel, EmailStr
from datetime import datetime
from typing import List, Optional

# User schemas
class UserBase(BaseModel):
    email: EmailStr
    username: str

class UserCreate(UserBase):
    password: str

class UserResponse(UserBase):
    id: int
    created_at: datetime
    
    class Config:
        from_attributes = True  # 允许从 ORM 模型创建

# Post schemas
class PostBase(BaseModel):
    title: str
    content: str

class PostCreate(PostBase):
    pass

class PostResponse(PostBase):
    id: int
    author_id: int
    created_at: datetime
    
    class Config:
        from_attributes = True

class UserWithPosts(UserResponse):
    posts: List[PostResponse] = []

实验 4：CRUD 路由

创建 main.py：

from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy.orm import Session
from typing import List

from database import engine, get_db, Base
from models import User, Post
from schemas import UserCreate, UserResponse, PostCreate, PostResponse, UserWithPosts

# 创建数据库表
Base.metadata.create_all(bind=engine)

app = FastAPI()

# User CRUD
@app.post("/users/", response_model=UserResponse, status_code=201)
def create_user(user: UserCreate, db: Session = Depends(get_db)):
    # 检查邮箱是否存在
    if db.query(User).filter(User.email == user.email).first():
        raise HTTPException(status_code=400, detail="Email already registered")
    
    # 创建用户（实际应用中应该哈希密码）
    db_user = User(
        email=user.email,
        username=user.username,
        hashed_password=user.password  # 简化示例，实际应使用 bcrypt
    )
    db.add(db_user)
    db.commit()
    db.refresh(db_user)
    return db_user

@app.get("/users/", response_model=List[UserResponse])
def list_users(skip: int = 0, limit: int = 10, db: Session = Depends(get_db)):
    users = db.query(User).offset(skip).limit(limit).all()
    return users

@app.get("/users/{user_id}", response_model=UserWithPosts)
def get_user(user_id: int, db: Session = Depends(get_db)):
    user = db.query(User).filter(User.id == user_id).first()
    if not user:
        raise HTTPException(status_code=404, detail="User not found")
    return user

# Post CRUD
@app.post("/users/{user_id}/posts/", response_model=PostResponse, status_code=201)
def create_post(user_id: int, post: PostCreate, db: Session = Depends(get_db)):
    # 验证用户存在
    if not db.query(User).filter(User.id == user_id).first():
        raise HTTPException(status_code=404, detail="User not found")
    
    db_post = Post(**post.dict(), author_id=user_id)
    db.add(db_post)
    db.commit()
    db.refresh(db_post)
    return db_post

@app.get("/posts/", response_model=List[PostResponse])
def list_posts(skip: int = 0, limit: int = 10, db: Session = Depends(get_db)):
    posts = db.query(Post).offset(skip).limit(limit).all()
    return posts

实验 5：测试完整流程

# 启动应用
uvicorn main:app --reload

# 创建用户
curl -X POST http://localhost:8000/users/ \
  -H "Content-Type: application/json" \
  -d '{"email":"john@example.com","username":"john","password":"secret"}'

# 输出: {"id":1,"email":"john@example.com","username":"john","created_at":"..."}

# 获取用户列表
curl http://localhost:8000/users/

# 创建文章
curl -X POST http://localhost:8000/users/1/posts/ \
  -H "Content-Type: application/json" \
  -d '{"title":"My First Post","content":"Hello World"}'

# 获取用户及其文章
curl http://localhost:8000/users/1

验收标准

✅ 数据库连接成功
✅ 用户 CRUD 操作正常
✅ 文章 CRUD 操作正常
✅ 用户和文章的关系正确
✅ 数据持久化到数据库

返回第三章概览