Day 18 | FastAPI 数据库整合

#!/usr/bin/env python3
"""
Day 18 - 数据库配置和模型定义
演示如何使用 SQLAlchemy 异步引擎和模型定义
"""

from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker, declarative_base
from sqlalchemy import Column, Integer, String, DateTime
from datetime import datetime

# ========== 1. 连接字符串 ==========
# 格式: mysql+aiomysql://用户名:密码@地址:端口/库名
# 重点: 必须明确指定 +aiomysql（表示使用异步驱动）
# 如果是 PostgreSQL: postgresql+asyncpg://...
DATABASE_URL = "mysql+aiomysql://root:root@localhost/infra_db"

# ========== 2. 创建异步引擎 ==========
# echo=True: 会打印生成的 SQL 语句，方便调试（生产环境应该设为 False）
# pool_pre_ping=True: 连接池预检查，自动重连断开的连接
engine = create_async_engine(
    DATABASE_URL, 
    echo=True,
    pool_pre_ping=True
)

# ========== 3. 创建会话工厂 ==========
# 以后每次请求进来，都从这里拿一个新的 session
# class_=AsyncSession: 指定使用异步会话
# expire_on_commit=False: 提交后不使对象过期（便于返回数据）
AsyncSessionLocal = sessionmaker(
    engine, 
    class_=AsyncSession, 
    expire_on_commit=False
)

# ========== 4. 声明基类 ==========
# 所有的 Model 都要继承这个 Base
Base = declarative_base()

# ========== 5. 定义 User 模型 ==========
class User(Base):
    """
    用户模型
    对应数据库中的 users 表
    """
    __tablename__ = "users"  # 对应数据库里的表名
    
    # 主键：自增整数
    id = Column(Integer, primary_key=True, index=True)
    
    # 用户名：字符串，唯一，非空
    username = Column(String(50), unique=True, nullable=False)
    
    # 邮箱：字符串，可为空
    email = Column(String(100), nullable=True)
    
    # 创建时间：日期时间，默认当前时间
    created_at = Column(DateTime, default=datetime.utcnow)

#!/usr/bin/env python3
"""
Day 18 - FastAPI 数据库整合实战
演示如何使用异步 SQLAlchemy 进行数据库操作
"""

from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy.future import select
from sqlalchemy.ext.asyncio import AsyncSession
from database import AsyncSessionLocal, User, Base, engine

# 初始化 FastAPI 应用
app = FastAPI(
    title="Database Integration API",
    description="演示 FastAPI + SQLAlchemy 异步数据库操作",
    version="1.0.0"
)

# ========== 启动时自动建表 ==========
# 仅用于开发环境，生产环境请使用 Alembic 进行数据库迁移
@app.on_event("startup")
async def startup():
    """
    应用启动时自动创建数据库表
    """
    async with engine.begin() as conn:
        # 创建所有继承自 Base 的模型对应的表
        await conn.run_sync(Base.metadata.create_all)
    print("✅ Database tables created")

# ========== 关键依赖: get_db ==========
# 这是一个 Generator 用于依赖注入
# 它的作用是：请求来了 -> 开连接 -> 给路由用 -> 路由跑完 -> 关连接
async def get_db():
    """
    获取数据库会话的依赖函数
    
    使用 Generator 模式，确保会话在使用完后自动关闭
    """
    async with AsyncSessionLocal() as session:
        # yield 表示这是一个 Generator
        # FastAPI 会在路由函数执行完后自动关闭 session
        yield session

# ========== 1. 创建用户 (POST) ==========
@app.post("/users/", status_code=201)
async def create_user(
    username: str,
    email: str = None,
    db: AsyncSession = Depends(get_db)
):
    """
    创建新用户
    
    参数:
        username: 用户名（必填）
        email: 邮箱（可选）
        db: 数据库会话（依赖注入）
    
    返回:
        User: 创建的用户对象
    """
    # 1. 检查用户名是否已存在
    result = await db.execute(
        select(User).where(User.username == username)
    )
    existing_user = result.scalars().first()
    
    if existing_user:
        raise HTTPException(
            status_code=400,
            detail="Username already exists"
        )
    
    # 2. 实例化 ORM 对象（此时还没进数据库）
    new_user = User(username=username, email=email)
    
    # 3. 添加到暂存区
    db.add(new_user)
    
    # 4. 提交事务（这一步才真正执行 SQL INSERT）
    # 注意: 必须加 await，因为这是网络 IO
    await db.commit()
    
    # 5. 刷新对象（为了拿回数据库生成的 id 和 created_at）
    await db.refresh(new_user)
    
    return new_user

# ========== 2. 查询用户列表 (GET) ==========
@app.get("/users/")
async def read_users(
    skip: int = 0,
    limit: int = 10,
    db: AsyncSession = Depends(get_db)
):
    """
    分页查询用户列表
    
    参数:
        skip: 跳过多少条记录（默认 0）
        limit: 返回多少条记录（默认 10）
        db: 数据库会话（依赖注入）
    
    返回:
        List[User]: 用户列表
    """
    # 新版 SQLAlchemy 查询语法: select(Model)
    # offset(skip): 跳过 skip 条
    # limit(limit): 只取 limit 条
    result = await db.execute(
        select(User).offset(skip).limit(limit)
    )
    
    # scalars().all() 把结果集转换成纯净的列表
    users = result.scalars().all()
    return users

# ========== 3. 查询单个用户 (GET) ==========
@app.get("/users/{user_id}")
async def read_user(
    user_id: int,
    db: AsyncSession = Depends(get_db)
):
    """
    查询单个用户
    
    参数:
        user_id: 用户 ID（路径参数）
        db: 数据库会话（依赖注入）
    
    返回:
        User: 用户对象
    """
    result = await db.execute(
        select(User).where(User.id == user_id)
    )
    user = result.scalars().first()
    
    if not user:
        raise HTTPException(
            status_code=404,
            detail="User not found"
        )
    
    return user

# ========== 4. 更新用户 (PUT) ==========
@app.put("/users/{user_id}")
async def update_user(
    user_id: int,
    email: str = None,
    db: AsyncSession = Depends(get_db)
):
    """
    更新用户信息
    
    参数:
        user_id: 用户 ID（路径参数）
        email: 新邮箱（可选）
        db: 数据库会话（依赖注入）
    
    返回:
        User: 更新后的用户对象
    """
    # 1. 查询用户
    result = await db.execute(
        select(User).where(User.id == user_id)
    )
    user = result.scalars().first()
    
    if not user:
        raise HTTPException(
            status_code=404,
            detail="User not found"
        )
    
    # 2. 更新字段
    if email is not None:
        user.email = email
    
    # 3. 提交事务
    await db.commit()
    await db.refresh(user)
    
    return user

# ========== 5. 删除用户 (DELETE) ==========
@app.delete("/users/{user_id}", status_code=204)
async def delete_user(
    user_id: int,
    db: AsyncSession = Depends(get_db)
):
    """
    删除用户
    
    参数:
        user_id: 用户 ID（路径参数）
        db: 数据库会话（依赖注入）
    
    返回:
        204 No Content（无响应体）
    """
    # 1. 查询用户
    result = await db.execute(
        select(User).where(User.id == user_id)
    )
    user = result.scalars().first()
    
    if not user:
        raise HTTPException(
            status_code=404,
            detail="User not found"
        )
    
    # 2. 删除用户
    await db.delete(user)
    
    # 3. 提交事务
    await db.commit()
    
    # 204 状态码表示成功但无响应体
    return None