第五十篇：SQLAlchemy——ORM 框架入门

前言

上一篇我们用原生 SQL 操作数据库，需要手写大量 SQL 语句并处理参数绑定、结果映射等细节。ORM（Object-Relational Mapping，对象关系映射） 把数据库表映射为 Python 类，把记录映射为对象，让你用面向对象的方式操作数据库。SQLAlchemy 是 Python 生态中最强大、最成熟的 ORM 框架，本文将系统讲解它的核心用法。

ORM 概念与优势

原生 SQL vs ORM

┌─────────────────────────────────────────────────┐
│              原生 SQL 方式                        │
│                                                  │
│  cursor.execute(                                 │
│      "SELECT * FROM users WHERE age > ?",        │
│      (18,)                                       │
│  )                                               │
│  rows = cursor.fetchall()                        │
│  # 手动映射字段 → 对象                             │
│  users = [User(r[0], r[1], r[2]) for r in rows]  │
└─────────────────────────────────────────────────┘

┌─────────────────────────────────────────────────┐
│              ORM 方式                             │
│                                                  │
│  users = session.query(User)\                   │
│      .filter(User.age > 18)\                     │
│      .all()                                       │
│  # 自动映射为 User 对象列表                        │
└─────────────────────────────────────────────────┘

ORM 的核心优势：

优势	说明
面向对象	用类和对象操作数据库，不用写 SQL
自动防注入	参数化查询内建，杜绝 SQL 注入
数据库无关	切换数据库引擎只需改配置
关系映射	自动处理外键、一对多、多对多
连接池管理	内建连接池，高效复用连接
迁移支持	配合 Alembic 自动生成迁移脚本

ORM 的代价：

• 性能略低于手写 SQL（多了一层抽象）
• 复杂查询需要额外学习 ORM 语法
• 对于极致性能场景可能不够灵活

安装与连接

安装

pip install sqlalchemy
# SQLAlchemy 2.0+ 推荐配合以下工具
pip install pymysql    # MySQL 驱动
pip install alembic    # 数据库迁移工具

连接数据库

from sqlalchemy import create_engine

# SQLite
engine_sqlite = create_engine('sqlite:///app.db', echo=True)

# MySQL（需安装 PyMySQL）
engine_mysql = create_engine(
    'mysql+pymysql://root:password@localhost:3306/test_db?charset=utf8mb4',
    echo=True,           # 打印 SQL 日志
    pool_size=10,        # 连接池大小
    max_overflow=5,      # 允许超出连接池的连接数
    pool_recycle=3600,   # 连接回收周期（秒）
    pool_pre_ping=True   # 连接前检查是否有效
)

# PostgreSQL
# engine_pg = create_engine('postgresql+psycopg2://user:pass@localhost/mydb')

# echo=True 会打印所有 SQL 语句，开发调试时打开，生产环境关闭

Engine 与 Connection 的关系

┌──────────────────────────────────┐
│           Engine                 │
│  ┌────────────────────────────┐  │
│  │     Pool（连接池）          │  │
│  │  ┌────┐ ┌────┐ ┌────┐     │  │
│  │  │Conn│ │Conn│ │Conn│     │  │
│  │  └────┘ └────┘ └────┘     │  │
│  └────────────────────────────┘  │
│         ↓ 获取连接                │
│  ┌────────────────────────────┐  │
│  │     Connection             │  │
│  │  ┌──────────────────────┐ │  │
│  │  │   Transaction        │ │  │
│  │  │  ┌────────────────┐ │ │  │
│  │  │  │  SQL 操作      │ │ │  │
│  │  │  └────────────────┘ │ │  │
│  │  └──────────────────────┘ │  │
│  └────────────────────────────┘  │
└──────────────────────────────────┘

定义模型与创建表

声明式基类

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

# 创建声明式基类
Base = declarative_base()

# 定义模型类
class User(Base):
    __tablename__ = 'users'
    
    id = Column(Integer, primary_key=True, autoincrement=True)
    username = Column(String(50), unique=True, nullable=False, index=True)
    email = Column(String(100), unique=True, nullable=False)
    age = Column(Integer, default=0)
    is_active = Column(Boolean, default=True)
    created_at = Column(DateTime, default=datetime.now)
    
    def __repr__(self):
        return f"<User(id={self.id}, username='{self.username}', email='{self.email}')>"

class Post(Base):
    __tablename__ = 'posts'
    
    id = Column(Integer, primary_key=True, autoincrement=True)
    title = Column(String(200), nullable=False)
    content = Column(Text)
    views = Column(Integer, default=0)
    likes = Column(Integer, default=0)
    author_id = Column(Integer, ForeignKey('users.id'))
    published = Column(Boolean, default=False)
    created_at = Column(DateTime, default=datetime.now)
    
    # 关系映射
    author = relationship("User", backref="posts")
    
    def __repr__(self):
        return f"<Post(id={self.id}, title='{self.title}', views={self.views})>"

class Comment(Base):
    __tablename__ = 'comments'
    
    id = Column(Integer, primary_key=True, autoincrement=True)
    content = Column(Text, nullable=False)
    post_id = Column(Integer, ForeignKey('posts.id'))
    author_id = Column(Integer, ForeignKey('users.id'))
    created_at = Column(DateTime, default=datetime.now)
    
    post = relationship("Post", backref="comments")
    author = relationship("User", backref="comments")
    
    def __repr__(self):
        return f"<Comment(id={self.id}, content='{self.content[:20]}...')>"

# 创建所有表
engine = create_engine('sqlite:///blog.db', echo=False)
Base.metadata.create_all(engine)
print("✓ 所有表创建成功")

常用列类型速查

SQLAlchemy 类型	Python 类型	数据库类型
Integer	int	INT / INTEGER
Float	float	FLOAT / REAL
String(n)	str	VARCHAR(n)
Text	str	TEXT
Boolean	bool	BOOLEAN / TINYINT
DateTime	datetime	DATETIME / TIMESTAMP
Date	date	DATE
Time	time	TIME
Numeric(p, s)	Decimal	DECIMAL
LargeBinary	bytes	BLOB
JSON	dict/list	JSON

Session——ORM 的核心

创建 Session

from sqlalchemy.orm import sessionmaker

engine = create_engine('sqlite:///blog.db', echo=False)
Session = sessionmaker(bind=engine)
session = Session()

Session 生命周期

from contextlib import contextmanager

@contextmanager
def session_scope():
    """推荐的 Session 上下文管理器"""
    session = Session()
    try:
        yield session
        session.commit()
    except Exception as e:
        session.rollback()
        raise e
    finally:
        session.close()

# 使用
with session_scope() as session:
    user = User(username='admin', email='admin@example.com', age=25)
    session.add(user)
    # 退出 with 块时自动 commit

增删改查（CRUD）

增（Create）

from sqlalchemy.orm import sessionmaker

engine = create_engine('sqlite:///blog.db', echo=False)
Session = sessionmaker(bind=engine)
session = Session()

# 单条添加
user1 = User(username='张三', email='zhangsan@example.com', age=25)
session.add(user1)
session.commit()
print(f"✓ 添加用户: {user1}")  # id 已自动填充

# 批量添加
users = [
    User(username='李四', email='lisi@example.com', age=30),
    User(username='王五', email='wangwu@example.com', age=28),
    User(username='赵六', email='zhaoliu@example.com', age=35),
    User(username='钱七', email='qianqi@example.com', age=22),
]
session.add_all(users)
session.commit()
for u in users:
    print(f"  ✓ {u.username}: ID={u.id}")

# 添加关联数据
post1 = Post(title='Python ORM 入门', content='SQLAlchemy 很强大...', author_id=user1.id)
post2 = Post(title='数据库设计技巧', content='三范式是基础...', author_id=user1.id, views=50)
session.add_all([post1, post2])
session.commit()
print(f"✓ 添加 {2} 篇文章")

session.close()

查（Read）

from sqlalchemy import select

session = Session()

# ---------- 基本查询 ----------

# 查询所有
all_users = session.query(User).all()
print(f"总用户数: {len(all_users)}")

# 按主键查询
user = session.get(User, 1)
print(f"ID=1 的用户: {user}")

# 条件查询
adults = session.query(User).filter(User.age >= 25).all()
print(f"\n年龄 >= 25 的用户:")
for u in adults:
    print(f"  {u.username}: {u.age}岁")

# 多条件组合
result = session.query(User).filter(
    User.age >= 25,
    User.is_active == True
).all()

# ---------- 排序、限制、偏移 ----------

# 排序
oldest = session.query(User).order_by(User.age.desc()).all()
print(f"\n按年龄降序:")
for u in oldest:
    print(f"  {u.username}: {u.age}岁")

# 限制
top3 = session.query(User).order_by(User.age.desc()).limit(3).all()
print(f"\n年龄最大的3位:")
for u in top3:
    print(f"  {u.username}: {u.age}岁")

# 偏移（分页）
page, size = 1, 2
page_data = session.query(User)\
    .order_by(User.id)\
    .offset((page - 1) * size)\
    .limit(size)\
    .all()
print(f"\n第{page}页:")
for u in page_data:
    print(f"  {u.username}")

# ---------- 聚合查询 ----------

from sqlalchemy import func

# 计数
total = session.query(User).count()
print(f"\n总用户数: {total}")

# 统计
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}, "
      f"最小: {stats.min_age}, 最大: {stats.max_age}")

# 分组统计
groups = session.query(
    User.is_active,
    func.count(User.id).label('cnt')
).group_by(User.is_active).all()
for g in groups:
    status = "活跃" if g.is_active else "停用"
    print(f"  {status}: {g.cnt}人")

# ---------- 模糊查询与条件组合 ----------

from sqlalchemy import or_, and_, like, ilike

# 模糊查询
zhang = session.query(User).filter(User.username.like('张%')).all()
print(f"\n姓张的用户: {[u.username for u in zhang]}")

# OR 条件
result = session.query(User).filter(
    or_(
        User.username.like('张%'),
        User.age > 30
    )
).all()
print(f"姓张或年龄>30: {[(u.username, u.age) for u in result]}")

# IN 查询
names = ['张三', '李四', '王五']
users_in = session.query(User).filter(User.username.in_(names)).all()
print(f"指定名字: {[u.username for u in users_in]}")

# NOT IN
users_not_in = session.query(User).filter(~User.username.in_(names)).all()
print(f"不在指定列表: {[u.username for u in users_not_in]}")

# BETWEEN
mid_age = session.query(User).filter(User.age.between(25, 35)).all()
print(f"年龄25-35: {[(u.username, u.age) for u in mid_age]}")

session.close()

改（Update）

session = Session()

# 方式一：查出后修改
user = session.query(User).filter_by(username='赵六').first()
if user:
    user.age = 36
    user.email = 'new_zhaoliu@example.com'
    session.commit()
    print(f"✓ 更新赵六: 年龄={user.age}, 邮箱={user.email}")

# 方式二：批量更新（高效）
updated = session.query(Post).filter(
    Post.views == 0
).update({Post.views: 10}, synchronize_session=False)
session.commit()
print(f"✓ 批量更新 {updated} 篇文章的浏览量")

# 方式三：使用 update() 构造器
from sqlalchemy import update

stmt = update(Post).where(Post.likes == 0).values(likes=5)
result = session.execute(stmt)
session.commit()
print(f"✓ 执行语句更新 {result.rowcount} 条记录")

session.close()

删（Delete）

session = Session()

# 方式一：查出后删除
user = session.query(User).filter_by(username='钱七').first()
if user:
    session.delete(user)
    session.commit()
    print(f"✓ 删除用户: {user.username}")

# 方式二：批量删除
deleted = session.query(Post).filter(Post.views < 5).delete(synchronize_session=False)
session.commit()
print(f"✓ 批量删除 {deleted} 篇文章")

# 方式三：使用 delete() 构造器
from sqlalchemy import delete

stmt = delete(Comment).where(Comment.content.like('%垃圾%'))
result = session.execute(stmt)
session.commit()
print(f"✓ 语句删除 {result.rowcount} 条评论")

session.close()

关系映射

一对多关系

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

Base = declarative_base()

class Department(Base):
    __tablename__ = 'departments'
    
    id = Column(Integer, primary_key=True)
    name = Column(String(50), nullable=False)
    
    # backref 自动在 Employee 上添加 department 属性
    employees = relationship("Employee", backref='department', lazy='dynamic')
    
    def __repr__(self):
        return f"<Department({self.name})>"

class Employee(Base):
    __tablename__ = 'employees'
    
    id = Column(Integer, primary_key=True)
    name = Column(String(50), nullable=False)
    salary = Column(Integer)
    dept_id = Column(Integer, ForeignKey('departments.id'))
    
    def __repr__(self):
        return f"<Employee({self.name}, ¥{self.salary})>"

engine = create_engine('sqlite:///company.db', echo=False)
Base.metadata.create_all(engine)
Session = sessionmaker(bind=engine)
session = Session()

# 创建部门和员工
tech = Department(name='技术部')
sales = Department(name='销售部')
hr = Department(name='人事部')
session.add_all([tech, sales, hr])
session.commit()

employees = [
    Employee(name='张三', salary=20000, dept_id=tech.id),
    Employee(name='李四', salary=18000, dept_id=tech.id),
    Employee(name='王五', salary=15000, dept_id=sales.id),
    Employee(name='赵六', salary=12000, dept_id=sales.id),
    Employee(name='钱七', salary=10000, dept_id=hr.id),
]
session.add_all(employees)
session.commit()

# 查询：部门 → 员工
print("=== 部门员工列表 ===")
for dept in session.query(Department).all():
    # lazy='dynamic' 返回的是 Query 对象，可以继续过滤
    emps = dept.employees.all()
    total_salary = sum(e.salary for e in emps)
    print(f"\n{dept.name} ({len(emps)}人, 总薪资 ¥{total_salary}):")
    for emp in emps:
        print(f"  {emp.name}: ¥{emp.salary}")

# 查询：员工 → 部门
print("\n=== 员工所属部门 ===")
for emp in session.query(Employee).all():
    print(f"  {emp.name} → {emp.department.name}")

# 关联查询
print("\n=== 技术部薪资 > 18000 ===")
high_paid = session.query(Employee).join(Department)\
    .filter(Department.name == '技术部')\
    .filter(Employee.salary > 18000)\
    .all()
for emp in high_paid:
    print(f"  {emp.name}: ¥{emp.salary}")

session.close()

多对多关系

from sqlalchemy import Table

Base = declarative_base()

# 关联表
post_tags = Table('post_tags', Base.metadata,
    Column('post_id', Integer, ForeignKey('posts2.id'), primary_key=True),
    Column('tag_id', Integer, ForeignKey('tags.id'), primary_key=True)
)

class Post2(Base):
    __tablename__ = 'posts2'
    
    id = Column(Integer, primary_key=True)
    title = Column(String(200), nullable=False)
    
    tags = relationship("Tag", secondary=post_tags, backref="posts")
    
    def __repr__(self):
        return f"<Post2({self.title})>"

class Tag(Base):
    __tablename__ = 'tags'
    
    id = Column(Integer, primary_key=True)
    name = Column(String(50), unique=True, nullable=False)
    
    def __repr__(self):
        return f"<Tag({self.name})>"

engine = create_engine('sqlite:///tags.db', echo=False)
Base.metadata.create_all(engine)
Session = sessionmaker(bind=engine)
session = Session()

# 创建标签
python_tag = Tag(name='Python')
web_tag = Tag(name='Web')
db_tag = Tag(name='数据库')
ai_tag = Tag(name='AI')
session.add_all([python_tag, web_tag, db_tag, ai_tag])
session.commit()

# 创建文章并关联标签
post1 = Post2(title='Python Web 开发指南')
post1.tags = [python_tag, web_tag]

post2 = Post2(title='数据库性能优化')
post2.tags = [db_tag, python_tag]

post3 = Post2(title='AI 与机器学习')
post3.tags = [ai_tag, python_tag]

session.add_all([post1, post2, post3])
session.commit()

# 查询：文章 → 标签
print("=== 文章标签 ===")
for post in session.query(Post2).all():
    tag_names = [t.name for t in post.tags]
    print(f"  {post.title}: {tag_names}")

# 查询：标签 → 文章
print("\n=== 标签下的文章 ===")
for tag in session.query(Tag).all():
    post_titles = [p.title for p in tag.posts]
    print(f"  [{tag.name}]: {post_titles}")

# 按标签筛选文章
print("\n=== Python 标签的文章 ===")
python_posts = session.query(Post2)\
    .join(Post2.tags)\
    .filter(Tag.name == 'Python')\
    .all()
for p in python_posts:
    print(f"  {p.title}")

session.close()

关系加载策略

"""
lazy 加载策略：

1. lazy='select' (默认): 访问关系属性时发一条 SELECT
2. lazy='joined':       用 JOIN 一次性加载（适合一对多）
3. lazy='subquery':     用子查询一次性加载（适合多对多）
4. lazy='dynamic':     返回 Query 对象，可继续过滤
5. lazy='selectin':    用 IN 查询批量加载（推荐）
"""

# 示例：使用 joinedload
from sqlalchemy.orm import joinedload, selectinload

session = Session()

# N+1 问题演示
print("=== N+1 问题（低效）===")
# 先查所有部门（1条SQL），再逐个查员工（N条SQL）= N+1
for dept in session.query(Department).all():
    emps = dept.employees.all()
    # print(f"{dept.name}: {len(emps)}人")  # 每次触发一条 SQL

# 解决方案：joinedload
print("\n=== joinedload（高效）===")
depts = session.query(Department)\
    .options(joinedload(Department.employees))\
    .all()
for dept in depts:
    print(f"  {dept.name}: {[e.name for e in dept.employees]}")

# 解决方案：selectinload（更适合一对多）
print("\n=== selectinload（推荐）===")
depts = session.query(Department)\
    .options(selectinload(Department.employees))\
    .all()
for dept in depts:
    print(f"  {dept.name}: {len(dept.employees.all())}人")

session.close()

SQLAlchemy 2.0 新语法

from sqlalchemy import create_engine, select
from sqlalchemy.orm import declarative_base, sessionmaker

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)
    stock = Column(Integer, default=0)
    category = Column(String(50))

engine = create_engine('sqlite:///shop.db', echo=False)
Base.metadata.create_all(engine)
Session = sessionmaker(bind=engine)
session = Session()

# 插入数据
session.add_all([
    Product(name='键盘', price=299, stock=50, category='电子'),
    Product(name='鼠标', price=89, stock=200, category='电子'),
    Product(name='杯子', price=39, stock=300, category='日用'),
])
session.commit()

# SQLAlchemy 2.0 推荐使用 select() 语法
print("=== 2.0 select 语法 ===")

# 查询所有
stmt = select(Product)
products = session.execute(stmt).scalars().all()
for p in products:
    print(f"  {p.name}: ¥{p.price}")

# 条件查询
stmt = select(Product).where(Product.price > 100).order_by(Product.price.desc())
expensive = session.execute(stmt).scalars().all()
print(f"\n价格 > 100: {[(p.name, p.price) for p in expensive]}")

# 聚合查询
from sqlalchemy import func
stmt = select(
    Product.category,
    func.count(Product.id).label('count'),
    func.avg(Product.price).label('avg_price')
).group_by(Product.category)
results = session.execute(stmt)
for row in results:
    print(f"  {row.category}: {row.count}个, 均价¥{row.avg_price:.2f}")

# 指定列查询（不返回完整对象）
stmt = select(Product.name, Product.price).where(Product.stock > 50)
rows = session.execute(stmt)
print("\n库存>50的商品:")
for name, price in rows:
    print(f"  {name}: ¥{price}")

session.close()

实战项目：博客系统数据层

from sqlalchemy import create_engine, Column, Integer, String, Text, DateTime, Boolean, ForeignKey, func
from sqlalchemy.orm import declarative_base, relationship, sessionmaker
from contextlib import contextmanager
from datetime import datetime, timedelta

Base = declarative_base()

class Author(Base):
    __tablename__ = 'blog_authors'
    
    id = Column(Integer, primary_key=True)
    name = Column(String(50), nullable=False)
    email = Column(String(100), unique=True)
    is_admin = Column(Boolean, default=False)
    created_at = Column(DateTime, default=datetime.now)
    
    articles = relationship("Article", backref="author", lazy="dynamic")
    
    def __repr__(self):
        return f"<Author({self.name})>"

class Article(Base):
    __tablename__ = 'blog_articles'
    
    id = Column(Integer, primary_key=True)
    title = Column(String(200), nullable=False)
    content = Column(Text)
    status = Column(String(20), default='draft')  # draft / published
    views = Column(Integer, default=0)
    likes = Column(Integer, default=0)
    author_id = Column(Integer, ForeignKey('blog_authors.id'))
    created_at = Column(DateTime, default=datetime.now)
    updated_at = Column(DateTime, default=datetime.now, onupdate=datetime.now)
    
    def __repr__(self):
        return f"<Article({self.title})>"

class BlogDB:
    """博客系统数据库操作封装"""
    
    def __init__(self, db_url='sqlite:///blog_system.db'):
        self.engine = create_engine(db_url, echo=False)
        Base.metadata.create_all(self.engine)
        self.Session = sessionmaker(bind=self.engine)
    
    @contextmanager
    def _session(self):
        session = self.Session()
        try:
            yield session
            session.commit()
        except Exception as e:
            session.rollback()
            raise e
        finally:
            session.close()
    
    # ---------- 作者管理 ----------
    
    def add_author(self, name, email, is_admin=False):
        with self._session() as s:
            author = Author(name=name, email=email, is_admin=is_admin)
            s.add(author)
            s.flush()
            return author.id
    
    def get_author(self, author_id):
        with self._session() as s:
            return s.get(Author, author_id)
    
    # ---------- 文章管理 ----------
    
    def add_article(self, title, content, author_id, status='draft'):
        with self._session() as s:
            article = Article(
                title=title, content=content,
                author_id=author_id, status=status
            )
            s.add(article)
            s.flush()
            return article.id
    
    def publish_article(self, article_id):
        with self._session() as s:
            article = s.get(Article, article_id)
            if article:
                article.status = 'published'
                return True
            return False
    
    def get_article(self, article_id):
        with self._session() as s:
            return s.get(Article, article_id)
    
    def get_published_articles(self, page=1, size=10):
        with self._session() as s:
            offset = (page - 1) * size
            return s.query(Article)\
                .filter(Article.status == 'published')\
                .order_by(Article.created_at.desc())\
                .offset(offset)\
                .limit(size)\
                .all()
    
    def search_articles(self, keyword):
        with self._session() as s:
            return s.query(Article)\
                .filter(Article.title.like(f'%{keyword}%'))\
                .order_by(Article.views.desc())\
                .all()
    
    def increment_views(self, article_id):
        with self._session() as s:
            article = s.get(Article, article_id)
            if article:
                article.views += 1
                return article.views
            return None
    
    def like_article(self, article_id):
        with self._session() as s:
            article = s.get(Article, article_id)
            if article:
                article.likes += 1
                return article.likes
            return None
    
    # ---------- 统计分析 ----------
    
    def get_author_stats(self):
        """作者文章统计"""
        with self._session() as s:
            return s.query(
                Author.name,
                func.count(Article.id).label('article_count'),
                func.sum(Article.views).label('total_views'),
                func.sum(Article.likes).label('total_likes')
            ).outerjoin(Article, Article.author_id == Author.id)\
             .group_by(Author.id)\
             .order_by(func.sum(Article.views).desc())\
             .all()
    
    def get_popular_articles(self, limit=5):
        """热门文章"""
        with self._session() as s:
            return s.query(Article)\
                .filter(Article.status == 'published')\
                .order_by(Article.views.desc())\
                .limit(limit)\
                .all()
    
    def get_recent_articles_by_author(self, author_id, days=7):
        """作者最近文章"""
        with self._session() as s:
            since = datetime.now() - timedelta(days=days)
            return s.query(Article)\
                .filter(
                    Article.author_id == author_id,
                    Article.created_at >= since
                )\
                .order_by(Article.created_at.desc())\
                .all()


# ===== 运行博客系统 =====
if __name__ == '__main__':
    db = BlogDB()
    
    # 创建作者
    a1 = db.add_author("鲁迅", "luxun@example.com", is_admin=True)
    a2 = db.add_author("老舍", "laoshe@example.com")
    a3 = db.add_author("冰心", "bingxin@example.com")
    
    # 创建文章
    articles_data = [
        ("狂人日记", "今天晚上，很好的月光...", a1, 'published'),
        ("阿Q正传", "阿Q不独是姓名...", a1, 'published'),
        ("骆驼祥子", "祥子来自农村...", a2, 'published'),
        ("茶馆", "这是茶馆...", a2, 'published'),
        ("繁星", "繁星闪烁着...", a3, 'published'),
        ("春水", "春水流淌...", a3, 'draft'),
    ]
    
    article_ids = []
    for title, content, author_id, status in articles_data:
        aid = db.add_article(title, content, author_id, status)
        article_ids.append(aid)
    
    print("✓ 创建完成\n")
    
    # 发布草稿
    db.publish_article(article_ids[-1])
    print("✓ 发布《春水》\n")
    
    # 模拟浏览和点赞
    for aid in article_ids:
        db.increment_views(aid)
        db.increment_views(aid)
        db.like_article(aid)
    
    # 查看热门文章
    print("=== 热门文章 ===")
    for article in db.get_popular_articles(3):
        print(f"  [{article.id}] {article.title} - 浏览{article.views}, 点赞{article.likes}")
    
    # 搜索文章
    print("\n=== 搜索 '春' ===")
    for article in db.search_articles('春'):
        print(f"  {article.title}")
    
    # 作者统计
    print("\n=== 作者统计 ===")
    for name, count, views, likes in db.get_author_stats():
        v = views or 0
        l = likes or 0
        print(f"  {name}: {count}篇, {v}浏览, {l}点赞")
    
    # 分页查询
    print("\n=== 已发布文章（第1页）===")
    for article in db.get_published_articles(page=1, size=3):
        print(f"  [{article.id}] {article.title} ({article.status})")

小结

知识点	说明
create_engine()	创建数据库引擎，配置连接池
declarative_base()	声明式基类，所有模型继承它
Column()	定义列类型、约束、索引
relationship()	定义关系映射（一对多、多对多）
sessionmaker	Session 工厂，管理数据库会话
session.add/add_all	添加对象到会话
session.query()	构建查询，支持链式过滤
filter()	条件过滤，类似 SQL WHERE
joinedload/selectinload	预加载策略，解决 N+1 问题
select() (2.0)	新版推荐查询语法，更清晰
func	聚合函数（count、sum、avg 等）

核心原则：ORM 不是为了取代 SQL，而是让你在大多数场景下更高效地操作数据库。遇到复杂查询，session.execute(text("SQL")) 随时可以回退到原生 SQL。

练习题

1. 使用 SQLAlchemy 定义一个「学生-课程-选课」三表模型（多对多关系），实现学生选课、退课、查询课表、统计选课人数等功能。

2. 实现一个分页查询函数，接收页码和每页条数，返回文章列表和总页数，并解决 N+1 查询问题。

3. 对比原生 SQL 和 SQLAlchemy ORM 在批量插入 1000 条记录时的性能差异，分析 ORM 的性能瓶颈在哪里，以及如何优化（如 bulk_save_objects、session.bulk_insert_mappings 等）。