🐍 Python Day49：ORM 进阶 — 一对多、多对多与复杂查询

🕐 预计用时：2-3 小时 ｜ 🎯 目标：掌握关系映射、查询过滤、排序、聚合

📖 今日目录

1. 一对多关系
2. 多对多关系
3. 关系查询
4. 高级过滤
5. 聚合查询
6. 子查询
7. 分组与 HAVING
8. 实战：博客系统数据模型
9. 今日小结

1. 一对多关系

一个作者有多篇文章，一篇文章属于一个作者 → 一对多。

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

Base = declarative_base()

class Author(Base):
    __tablename__ = "authors"

    id = Column(Integer, primary_key=True)
    name = Column(String(100), nullable=False)
    email = Column(String(200))

    # 一对多关系：一个作者有多篇文章
    articles = relationship("Article", back_populates="author", lazy="dynamic")

    def __repr__(self):
        return f"<Author(name='{self.name}')>"

class Article(Base):
    __tablename__ = "articles"

    id = Column(Integer, primary_key=True)
    title = Column(String(200), nullable=False)
    content = Column(Text)
    created_at = Column(DateTime, default=datetime.now)

    # 外键
    author_id = Column(Integer, ForeignKey("authors.id"), nullable=False)

    # 反向关系
    author = relationship("Author", back_populates="articles")

    def __repr__(self):
        return f"<Article(title='{self.title}')>"

# 创建引擎和表
engine = create_engine("sqlite:///blog.db", echo=False)
Base.metadata.create_all(engine)
Session = sessionmaker(bind=engine)

# 插入数据
with Session() as session:
    author = Author(name="张三", email="zhangsan@example.com")
    session.add(author)
    session.flush()  # 获取 author.id

    articles = [
        Article(title="Python 入门", content="学习 Python 很简单", author_id=author.id),
        Article(title="SQLAlchemy 教程", content="ORM 很强大", author_id=author.id),
        Article(title="Flask 实战", content="Web 开发入门", author_id=author.id),
    ]
    session.add_all(articles)
    session.commit()

    # 通过关系访问
    print(f"作者: {author.name}")
    for a in author.articles:
        print(f"  📝 {a.title}")

    # 反向访问
    article = articles[0]
    print(f"\n文章 '{article.title}' 的作者: {article.author.name}")

💡 relationship 的 lazy 参数：
• lazy="select"（默认）：访问时才加载
• lazy="joined"：JOIN 查询一起加载
• lazy="subquery"：子查询加载
• lazy="dynamic"：返回查询对象（可链式过滤）

2. 多对多关系

一篇文章有多个标签，一个标签对应多篇文章 → 多对多。

from sqlalchemy import Column, Integer, String, Table, ForeignKey
from sqlalchemy.orm import relationship

# 多对多需要一个中间表
article_tags = Table(
    "article_tags",
    Base.metadata,
    Column("article_id", Integer, ForeignKey("articles.id"), primary_key=True),
    Column("tag_id", Integer, ForeignKey("tags.id"), primary_key=True)
)

class Tag(Base):
    __tablename__ = "tags"

    id = Column(Integer, primary_key=True)
    name = Column(String(50), unique=True, nullable=False)

    # 多对多关系
    articles = relationship("Article", secondary=article_tags, back_populates="tags")

    def __repr__(self):
        return f"<Tag(name='{self.name}')>"

# 在 Article 模型中添加反向关系
# Article.tags = relationship("Tag", secondary=article_tags, back_populates="articles")

Base.metadata.create_all(engine)

# 使用
with Session() as session:
    # 创建标签
    tags = [
        Tag(name="Python"),
        Tag(name="Web"),
        Tag(name="数据库"),
        Tag(name="入门"),
    ]
    session.add_all(tags)
    session.flush()

    # 给文章添加标签
    article = session.query(Article).first()
    article.tags = [tags[0], tags[3]]  # Python, 入门
    session.commit()

    # 查询
    print(f"文章 '{article.title}' 的标签:")
    for tag in article.tags:
        print(f"  🏷️ {tag.name}")

    # 反向查询：某个标签下的所有文章
    python_tag = session.query(Tag).filter_by(name="Python").first()
    print(f"\n'Python' 标签的文章:")
    for a in python_tag.articles:
        print(f"  📝 {a.title}")

3. 关系查询

with Session() as session:
    # 通过关系查询
    author = session.query(Author).filter_by(name="张三").first()

    # 获取作者的所有文章（通过 relationship）
    articles = author.articles.all()  # lazy="dynamic" 时用 .all()
    print(f"张三有 {len(articles)} 篇文章")

    # 用 join 查询：找出写过 "Python" 标签文章的作者
    authors = session.query(Author).join(Article).join(Article.tags).filter(
        Tag.name == "Python"
    ).all()
    print(f"写过 Python 文章的作者: {[a.name for a in authors]}")

    # 按关系条件过滤
    # 找出有 3 篇以上文章的作者
    from sqlalchemy import func
    prolific = session.query(Author).join(Article).group_by(Author.id).having(
        func.count(Article.id) >= 3
    ).all()
    print(f"多产作者: {[a.name for a in prolific]}")

4. 高级过滤

from sqlalchemy import and_, or_, not_

with Session() as session:
    # AND 条件
    results = session.query(Article).filter(
        and_(
            Article.title.like("%Python%"),
            Article.created_at >= datetime(2024, 1, 1)
        )
    ).all()

    # OR 条件
    results = session.query(Article).filter(
        or_(
            Article.title.like("%Python%"),
            Article.title.like("%Flask%")
        )
    ).all()

    # NOT 条件
    results = session.query(Article).filter(
        not_(Article.title.like("%测试%"))
    ).all()

    # IN 查询
    results = session.query(Article).filter(
        Article.author_id.in_([1, 2, 3])
    ).all()

    # BETWEEN
    results = session.query(Article).filter(
        Article.created_at.between(datetime(2024, 1, 1), datetime(2024, 12, 31))
    ).all()

    # 排序
    results = session.query(Article).order_by(Article.created_at.desc()).all()

    # 多字段排序
    results = session.query(Article).order_by(
        Article.author_id.asc(),
        Article.created_at.desc()
    ).all()

    # 去重
    results = session.query(Article.author_id).distinct().all()

5. 聚合查询

from sqlalchemy import func

with Session() as session:
    # COUNT
    count = session.query(func.count(Article.id)).scalar()
    print(f"总文章数: {count}")

    # SUM
    total_score = session.query(func.sum(Score.score)).filter(Score.student_id == 1).scalar()
    print(f"张三总分: {total_score}")

    # AVG
    avg_score = session.query(func.avg(Score.score)).scalar()
    print(f"平均分: {avg_score:.1f}")

    # MAX / MIN
    max_score = session.query(func.max(Score.score)).scalar()
    min_score = session.query(func.min(Score.score)).scalar()
    print(f"最高分: {max_score}, 最低分: {min_score}")

    # 多个聚合
    stats = session.query(
        func.count(Score.id).label("total"),
        func.avg(Score.score).label("avg"),
        func.max(Score.score).label("max"),
        func.min(Score.score).label("min")
    ).first()
    print(f"统计: 总数={stats.total}, 平均={stats.avg:.1f}, 最高={stats.max}, 最低={stats.min}")

6. 子查询

with Session() as session:
    # 子查询：找出每个班级成绩最高的学生
    from sqlalchemy import func

    # 子查询：每个学生的总分
    subq = session.query(
        Score.student_id,
        func.sum(Score.score).label("total_score")
    ).group_by(Score.student_id).subquery()

    # 主查询：关联学生表
    results = session.query(
        Student.name,
        Student.class_name,
        subq.c.total_score
    ).join(subq, Student.id == subq.c.student_id).order_by(
        subq.c.total_score.desc()
    ).all()

    for name, cls, score in results:
        print(f"  {name} ({cls}): {score}分")

7. 分组与 HAVING

with Session() as session:
    # GROUP BY
    from sqlalchemy import func

    # 按作者统计文章数
    stats = session.query(
        Author.name,
        func.count(Article.id).label("article_count")
    ).join(Article).group_by(Author.id).all()

    for name, count in stats:
        print(f"  {name}: {count} 篇")

    # HAVING：筛选分组结果
    # 找出文章数 >= 2 的作者
    stats = session.query(
        Author.name,
        func.count(Article.id).label("count")
    ).join(Article).group_by(Author.id).having(
        func.count(Article.id) >= 2
    ).all()

    print(f"\n文章数 >= 2 的作者:")
    for name, count in stats:
        print(f"  {name}: {count} 篇")

8. 实战：博客系统数据模型

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

class User(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True)
    username = Column(String(50), unique=True, nullable=False)
    email = Column(String(200), unique=True)
    password_hash = Column(String(200))
    is_active = Column(Boolean, default=True)
    created_at = Column(DateTime, default=datetime.now)

    posts = relationship("Post", back_populates="author", lazy="dynamic")
    comments = relationship("Comment", back_populates="user", lazy="dynamic")

class Post(Base):
    __tablename__ = "posts"
    id = Column(Integer, primary_key=True)
    title = Column(String(200), nullable=False)
    content = Column(Text)
    published = Column(Boolean, default=False)
    created_at = Column(DateTime, default=datetime.now)
    updated_at = Column(DateTime, onupdate=datetime.now)

    author_id = Column(Integer, ForeignKey("users.id"))
    author = relationship("User", back_populates="posts")
    comments = relationship("Comment", back_populates="post", lazy="dynamic")
    tags = relationship("Tag", secondary="post_tags", back_populates="posts")

class Comment(Base):
    __tablename__ = "comments"
    id = Column(Integer, primary_key=True)
    content = Column(Text, nullable=False)
    created_at = Column(DateTime, default=datetime.now)

    user_id = Column(Integer, ForeignKey("users.id"))
    post_id = Column(Integer, ForeignKey("posts.id"))
    user = relationship("User", back_populates="comments")
    post = relationship("Post", back_populates="comments")

9. 今日小结

核心要点

• ✅ relationship() 定义 Python 层面的关系
• ✅ back_populates 双向关联
• ✅ 多对多需要中间表（Table + secondary）
• ✅ lazy="dynamic" 返回可链式过滤的查询对象
• ✅ func.count/sum/avg/max/min 聚合查询
• ✅ subquery() 子查询
• ✅ group_by().having() 分组筛选

🎯 练习建议：
1. 设计一个"电商"数据模型（用户/商品/订单/评价）
2. 实现"按标签查询文章"的多对多查询
3. 写出"每个作者最新一篇文章"的子查询

📚 Day49 完成！明天综合练习 — 个人记账本（Flask + SQLite）