🐍 第30课：数据库ORM

—— SQLAlchemy：Python ORM 之王

📌 本课目标

1️⃣ SQLAlchemy 核心

from sqlalchemy import create_engine, Column, Integer, String, Float, DateTime, ForeignKey
from sqlalchemy.orm import declarative_base, relationship, sessionmaker
from datetime import datetime

# 连接数据库
engine = create_engine("sqlite:///:memory:", echo=False)  # echo=True 查看 SQL
Base = declarative_base()

# 定义模型
class User(Base):
    __tablename__ = "users"
    
    id = Column(Integer, primary_key=True)
    name = Column(String(50), nullable=False)
    email = Column(String(100), unique=True, nullable=False)
    age = Column(Integer, default=0)
    created_at = Column(DateTime, default=datetime.now)
    
    # 关系
    posts = relationship("Post", back_populates="author", lazy="dynamic")
    
    def __repr__(self):
        return f"<User {self.name}>"

class Post(Base):
    __tablename__ = "posts"
    
    id = Column(Integer, primary_key=True)
    title = Column(String(200), nullable=False)
    content = Column(String(5000))
    author_id = Column(Integer, ForeignKey("users.id"))
    created_at = Column(DateTime, default=datetime.now)
    
    # 关系
    author = relationship("User", back_populates="posts")
    tags = relationship("Tag", secondary="post_tags", back_populates="posts")

class Tag(Base):
    __tablename__ = "tags"
    
    id = Column(Integer, primary_key=True)
    name = Column(String(50), unique=True, nullable=False)
    posts = relationship("Post", secondary="post_tags", back_populates="tags")

# 多对多关联表
from sqlalchemy import Table
post_tags = Table(
    "post_tags", Base.metadata,
    Column("post_id", Integer, ForeignKey("posts.id"), primary_key=True),
    Column("tag_id", Integer, ForeignKey("tags.id"), primary_key=True),
)

# 建表
Base.metadata.create_all(engine)

2️⃣ Session 与 CRUD

from sqlalchemy.orm import Session

# 创建 Session 工厂
SessionLocal = sessionmaker(bind=engine)

# CRUD 操作
def create_user(name, email, age=0):
    """创建用户"""
    with SessionLocal() as session:
        user = User(name=name, email=email, age=age)
        session.add(user)
        session.commit()
        session.refresh(user)  # 获取自增ID
        return user

def get_user(user_id):
    """查询用户"""
    with SessionLocal() as session:
        return session.query(User).filter(User.id == user_id).first()

def update_user(user_id, **kwargs):
    """更新用户"""
    with SessionLocal() as session:
        user = session.query(User).get(user_id)
        if user:
            for key, value in kwargs.items():
                setattr(user, key, value)
            session.commit()
            return True
    return False

def delete_user(user_id):
    """删除用户"""
    with SessionLocal() as session:
        user = session.query(User).get(user_id)
        if user:
            session.delete(user)
            session.commit()
            return True
    return False

# 测试
u1 = create_user("张三", "zhang@example.com", 25)
u2 = create_user("李四", "li@example.com", 30)
print(f"创建: {u1}, ID={u1.id}")
print(f"查询: {get_user(1)}")
update_user(1, age=26)
print(f"更新: {get_user(1)}")

3️⃣ 高级查询

from sqlalchemy import func, and_, or_, desc

with SessionLocal() as session:
    # 基本查询
    all_users = session.query(User).all()
    
    # 条件过滤
    young = session.query(User).filter(User.age < 30).all()
    
    # 多条件
    result = session.query(User).filter(
        and_(User.age >= 20, User.age <= 35)
    ).all()
    
    # 模糊查询
    zhang = session.query(User).filter(User.name.like("%张%")).all()
    
    # 排序
    by_age = session.query(User).order_by(desc(User.age)).all()
    
    # 限制
    top3 = session.query(User).limit(3).all()
    
    # 聚合
    avg_age = session.query(func.avg(User.age)).scalar()
    count = session.query(func.count(User.id)).scalar()
    
    # 分组
    from collections import defaultdict
    age_groups = session.query(User.age, func.count(User.id)).group_by(User.age).all()
    
    # 关系查询
    user = session.query(User).first()
    user_posts = user.posts.all()  # 一对多
    
    # join 查询
    results = session.query(User.name, Post.title).join(Post).all()

4️⃣ 验证脚本

#!/usr/bin/env python3
"""第30课 SQLAlchemy验证"""
from sqlalchemy import create_engine, Column, Integer, String, Float
from sqlalchemy.orm import declarative_base, sessionmaker

engine = create_engine("sqlite:///:memory:", echo=False)
Base = declarative_base()

class Product(Base):
    __tablename__ = "products"
    id = Column(Integer, primary_key=True)
    name = Column(String(100), nullable=False)
    price = Column(Float, nullable=False)

Base.metadata.create_all(engine)
Session = sessionmaker(bind=engine)

def test_orm_crud():
    with Session() as session:
        # Create
        p1 = Product(name="Book", price=29.9)
        p2 = Product(name="Pen", price=5.5)
        session.add_all([p1, p2])
        session.commit()
        assert p1.id is not None
        
        # Read
        found = session.query(Product).filter_by(name="Book").first()
        assert found is not None
        assert found.price == 29.9
        
        # Update
        found.price = 39.9
        session.commit()
        updated = session.query(Product).get(found.id)
        assert updated.price == 39.9
        
        # Delete
        session.delete(p2)
        session.commit()
        count = session.query(Product).count()
        assert count == 1
    
    print("✅ ORM CRUD测试通过")

def test_relationship():
    from sqlalchemy import Column, Integer, String, ForeignKey
    from sqlalchemy.orm import relationship
    
    class Author(Base):
        __tablename__ = "authors"
        id = Column(Integer, primary_key=True)
        name = Column(String(50))
        books = relationship("Book", back_populates="author")
    
    class Book(Base):
        __tablename__ = "books"
        id = Column(Integer, primary_key=True)
        title = Column(String(100))
        author_id = Column(Integer, ForeignKey("authors.id"))
        author = relationship("Author", back_populates="books")
    
    Base.metadata.create_all(engine)
    
    with Session() as session:
        author = Author(name="鲁迅")
        book1 = Book(title="呐喊", author=author)
        book2 = Book(title="彷徨", author=author)
        session.add_all([author, book1, book2])
        session.commit()
        
        assert len(author.books) == 2
        assert book1.author.name == "鲁迅"
    
    print("✅ 关系映射测试通过")

if __name__ == "__main__":
    test_orm_crud()
    test_relationship()
    print("\n🎉 第30课全部验证通过！")

5️⃣ 数据库迁移概念

"""Alembic 是 SQLAlchemy 的数据库迁移工具：
- 版本化管理表结构变更
- 支持升级/回退
- 团队协作时同步数据库结构

安装：pip install alembic
初始化：alembic init migrations
生成迁移：alembic revision --autogenerate -m "add users table"
执行迁移：alembic upgrade head
回退：alembic downgrade -1
"""

6️⃣ 连接池与性能

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, scoped_session
from sqlalchemy.pool import QueuePool

# 连接池配置
engine = create_engine(
    "sqlite:///myapp.db",
    poolclass=QueuePool,
    pool_size=5,          # 常驻连接数
    max_overflow=10,      # 最大溢出连接数
    pool_timeout=30,      # 获取连接超时(秒)
    pool_recycle=3600,    # 连接回收时间(秒)
    echo=False,           # 打印SQL(调试用)
)

# 线程安全的 Session 工厂
SessionFactory = sessionmaker(bind=engine)
Session = scoped_session(SessionFactory)

# 批量插入优化
def bulk_insert(session, model_class, data_list, batch_size=1000):
    for i in range(0, len(data_list), batch_size):
        batch = data_list[i:i+batch_size]
        session.bulk_insert_mappings(model_class, batch)
        session.commit()

7️⃣ 查询优化

from sqlalchemy import create_engine, Column, Integer, String, func, text
from sqlalchemy.orm import declarative_base, sessionmaker

engine = create_engine("sqlite:///:memory:", echo=False)
Base = declarative_base()

class User(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True)
    name = Column(String(50))
    email = Column(String(100))
    age = Column(Integer)

Base.metadata.create_all(engine)
Session = sessionmaker(bind=engine)

# 批量插入测试数据
with Session() as session:
    session.add_all([User(name=f"用户{i}", email=f"user{i}@example.com", age=20+i%30) 
                     for i in range(1000)])
    session.commit()

with Session() as session:
    # 1. 只查需要的列
    names = session.query(User.name).all()  # 比 session.query(User).all() 快
    
    # 2. 使用索引（创建索引加速 WHERE）
    # CREATE INDEX idx_email ON users(email)
    
    # 3. 分页
    page = 2
    per_page = 20
    users = session.query(User).offset((page-1)*per_page).limit(per_page).all()
    
    # 4. exists 判断
    from sqlalchemy import exists
    has_admin = session.query(exists().where(User.name == "用户0")).scalar()
    
    # 5. 原生 SQL
    result = session.execute(text("SELECT COUNT(*) FROM users WHERE age > :age"), {"age": 30})
    count = result.scalar()
    
    # 6. 聚合
    stats = session.query(
        func.count(User.id).label("total"),
        func.avg(User.age).label("avg_age"),
        func.min(User.age).label("min_age"),
        func.max(User.age).label("max_age"),
    ).first()
    
    print(f"统计: 总{stats.total}, 平均{stats.avg_age:.1f}岁, 范围{stats.min_age}-{stats.max_age}岁")

8️⃣ SQLAlchemy 2.0 新特性

"""SQLAlchemy 2.0 主要变化：
1. 全新的类型提示支持
2. 异步引擎和会话
3. Dataclass 集成
4. 更简洁的查询 API
"""

# SQLAlchemy 2.0 声明式映射
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
from sqlalchemy import String, Integer, Float

class Base(DeclarativeBase):
    pass

class Product(Base):
    __tablename__ = "products"
    
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(100))
    price: Mapped[float] = mapped_column(Float)
    stock: Mapped[int] = mapped_column(Integer, default=0)
    
    def __repr__(self):
        return f"Product({self.name}, ¥{self.price})"

# 2.0 风格查询
from sqlalchemy import select

# 替代 session.query()
stmt = select(Product).where(Product.price > 100).order_by(Product.price)
results = session.execute(stmt).scalars().all()

# 异步支持
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession

async_engine = create_async_engine("sqlite+aiosqlite:///app.db")
async with AsyncSession(async_engine) as session:
    result = await session.execute(select(Product))
    products = result.scalars().all()

🔑 本课要点