SQLAlchemy进阶：25个提升Python数据库操作效率的绝招

py2024年12月29日约 27 分钟

开篇：告别低效ORM，拥抱高效SQLAlchemy

是不是觉得每次操作数据库都要写一堆 SQL 语句很麻烦？是不是觉得 ORM 查询效率很低？别慌！ SQLAlchemy 可以帮你解决这些问题！它既能让你像写 Python 代码一样操作数据库，又能兼顾性能和灵活性。今天，我就带你深入 SQLAlchemy 的腹地，探索那些隐藏的 “绝招”，让你的数据库操作效率直接起飞！

1. 基础篇：理解 Session 的生命周期

痛点: 对 Session 的生命周期不熟悉，导致各种莫名其妙的问题。

方案: 掌握 Session 的创建、使用和关闭。

from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.orm import sessionmaker
from sqlalchemy.ext.declarative import declarative_base

# 定义数据库引擎
engine = create_engine('sqlite:///:memory:')

# 定义 ORM 基类
Base = declarative_base()

# 定义模型类
classUser(Base):
    __tablename__ = 'users'
    id = Column(Integer, primary_key=True)
    name = Column(String)

# 创建表
Base.metadata.create_all(engine)

# 创建 Session 类
Session = sessionmaker(bind=engine)

# 创建 Session 对象
session = Session()

try:
# 添加数据
    user = User(name='Alice')
    session.add(user)
    session.commit()

# 查询数据
    retrieved_user = session.query(User).filter_by(name='Alice').first()
    print(f"Retrieved user: {retrieved_user.name}")

except Exception as e:
    session.rollback()
    print(f"Error: {e}")

finally:
    session.close() # 确保 Session 在使用完后关闭

# 输出
# Retrieved user: Alice

解释: Session 是 SQLAlchemy 中用于操作数据库的核心对象；使用 session.add() 添加数据， session.query() 查询数据；使用 session.commit() 提交事务， session.rollback() 回滚事务; session.close() 关闭session。类比: Session 就像一个数据库连接的代理人，负责管理你的数据库事务。

小任务: 创建一个 Session 对象，并使用 session.add_all() 添加多条数据。

2. 使用 Declarative Base 定义模型

痛点: 每次都要手写 __table__ 很麻烦。

方案: 使用 Declarative Base 简化模型定义。

from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

engine = create_engine('sqlite:///:memory:')
Base = declarative_base()

classProduct(Base):
    __tablename__ = 'products'
    id = Column(Integer, primary_key=True)
    name = Column(String)
    price = Column(Integer)

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

product = Product(name='Laptop', price=1200)
session.add(product)
session.commit()

retrieved_product = session.query(Product).filter_by(name='Laptop').first()
print(f"Retrieved product: {retrieved_product.name}, Price: {retrieved_product.price}")

session.close()

# 输出
# Retrieved product: Laptop, Price: 1200

解释:declarative_base() 创建一个基类，简化模型定义，不用手写__table__；模型类通过继承这个基类获得 ORM 功能。类比: Declarative Base 就像一个模版，让你快速创建具有 ORM 功能的模型类。

小任务: 使用 Declarative Base 定义一个包含 ForeignKey 的模型类。

3. 使用 Relationships 处理关联关系

痛点: 处理一对多、多对多关系很复杂。

方案: 使用 relationship() 函数定义关联关系。

from sqlalchemy import create_engine, Column, Integer, String, ForeignKey
from sqlalchemy.orm import relationship, sessionmaker
from sqlalchemy.ext.declarative import declarative_base

engine = create_engine('sqlite:///:memory:')
Base = declarative_base()

classAuthor(Base):
    __tablename__ = 'authors'
    id = Column(Integer, primary_key=True)
    name = Column(String)
    books = relationship("Book", backref="author")

classBook(Base):
    __tablename__ = 'books'
    id = Column(Integer, primary_key=True)
    title = Column(String)
    author_id = Column(Integer, ForeignKey('authors.id'))

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

author = Author(name='Jane Austen')
book1 = Book(title='Pride and Prejudice', author=author)
book2 = Book(title='Sense and Sensibility', author=author)

session.add(author)
session.add(book1)
session.add(book2)
session.commit()

retrieved_author = session.query(Author).filter_by(name='Jane Austen').first()
print(f"Author: {retrieved_author.name}")
for book in retrieved_author.books:
    print(f"  Book: {book.title}")

session.close()

# 输出
# Author: Jane Austen
#   Book: Pride and Prejudice
#   Book: Sense and Sensibility

解释:relationship() 函数定义关联关系; backref 允许从 Book 访问 author; ForeignKey 定义外键关系。类比:relationship() 就像一条连接两个模型类的桥梁，让你可以轻松访问关联数据。

小任务: 创建一个多对多关系的示例，并使用 relationship() 和 secondary 参数进行定义。

4. 使用 Query 的各种过滤条件

痛点: 只知道简单的 filter_by，无法满足复杂查询需求。

方案: 掌握 filter 和各种 SQL 函数。

from sqlalchemy import create_engine, Column, Integer, String, and_, or_, not_
from sqlalchemy.orm import sessionmaker
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import func

engine = create_engine('sqlite:///:memory:')
Base = declarative_base()

classEmployee(Base):
    __tablename__ = 'employees'
    id = Column(Integer, primary_key=True)
    name = Column(String)
    department = Column(String)
    salary = Column(Integer)

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

session.add_all([
    Employee(name='Alice', department='IT', salary=6000),
    Employee(name='Bob', department='HR', salary=5000),
    Employee(name='Charlie', department='IT', salary=7000),
    Employee(name='David', department='Sales', salary=8000)
])
session.commit()


# 使用 filter 和 and_ 条件
employees = session.query(Employee).filter(and_(Employee.department == 'IT', Employee.salary > 6500)).all()
print("Employees in IT with salary > 6500:")
for emp in employees:
    print(f"- {emp.name}")

# 使用 or_ 条件
employees = session.query(Employee).filter(or_(Employee.department == 'HR', Employee.salary > 7000)).all()
print("\nEmployees in HR or with salary > 7000:")
for emp in employees:
    print(f"- {emp.name}")

# 使用 not_ 条件
employees = session.query(Employee).filter(not_(Employee.department == 'IT')).all()
print("\nEmployees not in IT:")
for emp in employees:
    print(f"- {emp.name}")


# 使用 SQL 函数
avg_salary = session.query(func.avg(Employee.salary)).scalar()
print(f"\nAverage salary: {avg_salary}")


session.close()

# 输出
# Employees in IT with salary > 6500:
# - Charlie
#
# Employees in HR or with salary > 7000:
# - Bob
# - David
#
# Employees not in IT:
# - Bob
# - David
#
# Average salary: 6500.0

解释: 使用 filter() 结合 and_(), or_(), not_() 实现复杂逻辑；func 模块提供 SQL 函数调用，如 func.avg()。类比:filter() 就像一个过滤器，可以根据你的条件，筛选出你需要的数据。

小任务: 查询工资在某个范围内的员工，并按照工资从高到低排序。

5. 使用 `limit` 和 `offset` 分页查询

痛点: 一次性查询大量数据，性能差。

方案: 使用 limit 和 offset 实现分页查询。

from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.orm import sessionmaker
from sqlalchemy.ext.declarative import declarative_base

engine = create_engine('sqlite:///:memory:')
Base = declarative_base()

classArticle(Base):
    __tablename__ = 'articles'
    id = Column(Integer, primary_key=True)
    title = Column(String)

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

session.add_all([Article(title=f'Article {i}') for i in range(1, 11)])
session.commit()

page_size = 3
page_num = 2

articles = session.query(Article).limit(page_size).offset((page_num - 1) * page_size).all()
print(f"Page {page_num} articles (page size: {page_size}):")
for article in articles:
    print(f"- {article.title}")

session.close()

# 输出
# Page 2 articles (page size: 3):
# - Article 4
# - Article 5
# - Article 6

解释:limit() 限制查询结果数量; offset() 指定查询起始位置，实现分页。类比:limit 和 offset 就像一本书的页码，可以让你跳到指定页面查看内容。

小任务: 封装一个分页查询的函数，接受页码和页大小作为参数。

6. 使用 `order_by` 进行排序

痛点: 查询结果没有按照预期的顺序排列。

方案: 使用 order_by 函数进行排序。

from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.orm import sessionmaker
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import desc

engine = create_engine('sqlite:///:memory:')
Base = declarative_base()

classTask(Base):
    __tablename__ = 'tasks'
    id = Column(Integer, primary_key=True)
    name = Column(String)
    priority = Column(Integer)

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

session.add_all([
    Task(name='Task 1', priority=2),
    Task(name='Task 2', priority=1),
    Task(name='Task 3', priority=3)
])
session.commit()

# 按优先级升序排列
tasks_asc = session.query(Task).order_by(Task.priority).all()
print("Tasks ordered by priority (ascending):")
for task in tasks_asc:
    print(f"- {task.name}, Priority: {task.priority}")

# 按优先级降序排列
tasks_desc = session.query(Task).order_by(desc(Task.priority)).all()
print("\nTasks ordered by priority (descending):")
for task in tasks_desc:
    print(f"- {task.name}, Priority: {task.priority}")

session.close()

# 输出
# Tasks ordered by priority (ascending):
# - Task 2, Priority: 1
# - Task 1, Priority: 2
# - Task 3, Priority: 3
#
# Tasks ordered by priority (descending):
# - Task 3, Priority: 3
# - Task 1, Priority: 2
# - Task 2, Priority: 1

解释:order_by() 函数进行排序；desc() 函数用于降序排序。类比:order_by() 就像一个排序器，可以按照指定的列和顺序对数据进行排序。

小任务: 使用多个列进行排序，先按照优先级降序排序，再按照名称升序排序。

7. 使用 `group_by` 和 `having` 进行分组统计

痛点: 需要对数据进行分组统计。

方案: 使用 group_by 和 having 函数进行分组统计。

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

engine = create_engine('sqlite:///:memory:')
Base = declarative_base()

classOrder(Base):
    __tablename__ = 'orders'
    id = Column(Integer, primary_key=True)
    product = Column(String)
    quantity = Column(Integer)

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

session.add_all([
    Order(product='Laptop', quantity=2),
    Order(product='Laptop', quantity=3),
    Order(product='Tablet', quantity=1),
    Order(product='Tablet', quantity=2),
    Order(product='Phone', quantity=5)
])
session.commit()

#  按产品分组，统计总数量
results = session.query(Order.product, func.sum(Order.quantity)).group_by(Order.product).all()
print("Total quantity by product:")
for product, total_quantity in results:
    print(f"- {product}: {total_quantity}")

# 按产品分组, 过滤总数量大于 3 的
results = session.query(Order.product, func.sum(Order.quantity)).group_by(Order.product).having(func.sum(Order.quantity) > 3).all()
print("\nTotal quantity by product (having quantity > 3):")
for product, total_quantity in results:
    print(f"- {product}: {total_quantity}")


session.close()

# 输出
# Total quantity by product:
# - Laptop: 5
# - Phone: 5
# - Tablet: 3
#
# Total quantity by product (having quantity > 3):
# - Laptop: 5
# - Phone: 5

解释:group_by() 用于分组; having() 用于对分组后的结果进行过滤; func.sum() 计算总和。类比:group_by() 就像一个分类器，可以根据指定的列，对数据进行分组； having() 就像一个过滤器，可以过滤出指定分组的数据。

小任务: 按部门分组，统计每个部门的平均工资，并过滤出平均工资大于 6000 的部门。

8. 使用 `join` 进行多表连接查询

痛点: 需要联合多个表进行查询。

方案: 使用 join() 函数进行多表连接查询。

from sqlalchemy import create_engine, Column, Integer, String, ForeignKey
from sqlalchemy.orm import relationship, sessionmaker
from sqlalchemy.ext.declarative import declarative_base

engine = create_engine('sqlite:///:memory:')
Base = declarative_base()

classCustomer(Base):
    __tablename__ = 'customers'
    id = Column(Integer, primary_
```markdown
key=True)
    name = Column(String)

classOrder(Base):
    __tablename__ = 'orders'
    id = Column(Integer, primary_key=True)
    customer_id = Column(Integer, ForeignKey('customers.id'))
    product = Column(String)
    customer = relationship("Customer", backref="orders")

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

customer1 = Customer(name='Alice')
customer2 = Customer(name='Bob')
session.add_all([customer1, customer2])

order1 = Order(customer=customer1, product='Laptop')
order2 = Order(customer=customer1, product='Tablet')
order3 = Order(customer=customer2, product='Phone')
session.add_all([order1, order2, order3])
session.commit()

# 使用 join 连接查询
results = session.query(Customer.name, Order.product).join(Order).all()
print("Customer and their orders:")
for name, product in results:
    print(f"- {name}: {product}")

#使用 join + filter
results = session.query(Customer.name, Order.product).join(Order).filter(Customer.name == 'Alice').all()
print("\nAlice and her orders:")
for name, product in results:
  print(f"- {name}: {product}")


session.close()


# 输出
# Customer and their orders:
# - Alice: Laptop
# - Alice: Tablet
# - Bob: Phone
#
# Alice and her orders:
# - Alice: Laptop
# - Alice: Tablet

解释:join() 函数连接两个表; relationship() 用于指定关联关系。类比:join() 就像一个连接器，将多个表连接在一起，让你查询关联数据。

小任务: 查询所有购买了 "Laptop" 产品的客户名称。

9. 使用 Subquery 进行子查询

痛点: 需要进行复杂的嵌套查询。

方案: 使用 subquery() 函数创建子查询。

from sqlalchemy import create_engine, Column, Integer, String, ForeignKey
from sqlalchemy.orm import sessionmaker
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import func

engine = create_engine('sqlite:///:memory:')
Base = declarative_base()

classDepartment(Base):
    __tablename__ = 'departments'
    id = Column(Integer, primary_key=True)
    name = Column(String)

classEmployee(Base):
    __tablename__ = 'employees'
    id = Column(Integer, primary_key=True)
    name = Column(String)
    department_id = Column(Integer, ForeignKey('departments.id'))
    salary = Column(Integer)

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

dept1 = Department(name='IT')
dept2 = Department(name='HR')
session.add_all([dept1, dept2])

session.add_all([
    Employee(name='Alice', department_id=dept1.id, salary=6000),
    Employee(name='Bob', department_id=dept2.id, salary=5000),
    Employee(name='Charlie', department_id=dept1.id, salary=7000),
    Employee(name='David', department_id=dept2.id, salary=8000)
])
session.commit()

# 创建子查询
subquery = session.query(Employee.department_id, func.avg(Employee.salary).label('avg_salary')).group_by(Employee.department_id).subquery()
# 使用子查询
results = session.query(Department.name, subquery.c.avg_salary).join(subquery, Department.id == subquery.c.department_id).all()

print("Department and their average salary:")
for name, avg_salary in results:
  print(f"- {name}: {avg_salary}")
session.close()


# 输出
# Department and their average salary:
# - IT: 6500.0
# - HR: 6500.0

解释:subquery() 函数创建子查询; 使用 label() 为子查询结果命名。类比:subquery() 就像一个嵌套的查询，可以先查询出部分数据，然后在外层查询中使用这些数据。

小任务: 查询工资高于所在部门平均工资的员工姓名。

10. 使用 `exists()` 进行存在性判断

痛点: 判断数据是否存在，但不想查询出具体数据。

方案: 使用 exists() 函数进行存在性判断。

from sqlalchemy import create_engine, Column, Integer, String, exists
from sqlalchemy.orm import sessionmaker
from sqlalchemy.ext.declarative import declarative_base

engine = create_engine('sqlite:///:memory:')
Base = declarative_base()

classUser(Base):
    __tablename__ = 'users'
    id = Column(Integer, primary_key=True)
    name = Column(String)

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

session.add(User(name='Alice'))
session.commit()

# 判断用户是否存在
exists_query = session.query(exists().where(User.name == 'Alice')).scalar()
print(f"User Alice exists: {exists_query}")
exists_query = session.query(exists().where(User.name == 'Bob')).scalar()
print(f"User Bob exists: {exists_query}")

session.close()

# 输出
# User Alice exists: True
# User Bob exists: False

解释:exists() 函数进行存在性判断，返回布尔值，提高查询效率。类比:exists() 就像一个“探测器”，告诉你数据是否存在，而不用返回具体内容。

小任务: 判断表中是否存在满足特定条件的记录。

11. 使用 `update()` 方法进行数据更新

痛点: 每次更新都要先查询再修改很麻烦。

方案: 使用 update() 方法进行数据更新。

from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.orm import sessionmaker
from sqlalchemy.ext.declarative import declarative_base

engine = create_engine('sqlite:///:memory:')
Base = declarative_base()

classProduct(Base):
    __tablename__ = 'products'
    id = Column(Integer, primary_key=True)
    name = Column(String)
    price = Column(Integer)

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

session.add(Product(name='Laptop', price=1200))
session.commit()

# 使用 update 更新数据
session.query(Product).filter_by(name='Laptop').update({Product.price: 1300})
session.commit()

retrieved_product = session.query(Product).filter_by(name='Laptop').first()
print(f"Updated price: {retrieved_product.price}")
session.close()


# 输出
# Updated price: 1300

解释:update() 方法更新数据库记录，避免先查询再更新的繁琐步骤。类比:update() 就像一个 “快捷修改器”，直接修改数据库中的数据。

小任务: 使用 update() 方法批量更新满足特定条件的数据。

12. 使用 `delete()` 方法删除数据

痛点: 每次删除都要先查询再删除很麻烦。

方案: 使用 delete() 方法删除数据。

from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.orm import sessionmaker
from sqlalchemy.ext.declarative import declarative_base

engine = create_engine('sqlite:///:memory:')
Base = declarative_base()

classUser(Base):
    __tablename__ = 'users'
    id = Column(Integer, primary_key=True)
    name = Column(String)

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

session.add(User(name='Alice'))
session.add(User(name='Bob'))
session.commit()

# 使用 delete 删除数据
session.query(User).filter_by(name='Alice').delete()
session.commit()

users = session.query(User).all()
print("Remaining users:")
for user in users:
    print(f"- {user.name}")
session.close()


# 输出
# Remaining users:
# - Bob

解释:delete() 方法删除数据库记录，避免先查询再删除的繁琐步骤。类比:delete() 就像一个 “清除器”，直接从数据库中删除数据。

小任务: 使用 delete() 方法批量删除满足特定条件的数据。

13. 使用 with 语句管理 Session 上下文

痛点: 忘记关闭 Session 导致资源泄露。

方案: 使用 with 语句自动管理 Session 上下文。

from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.orm import sessionmaker
from sqlalchemy.ext.declarative import declarative_base

engine = create_engine('sqlite:///:memory:')
Base = declarative_base()

classUser(Base):
    __tablename__ = 'users'
    id = Column(Integer, primary_key=True)
    name = Column(String)

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

with Session() as session:
    user = User(name='Alice')
    session.add(user)
    session.commit()

    retrieved_user = session.query(User).filter_by(name='Alice').first()
    print(f"Retrieved user: {retrieved_user.name}")

# 输出
# Retrieved user: Alice

解释: 使用 with 语句，Session 会在代码块结束时自动关闭，避免资源泄露。类比:with 语句就像一个“保险箱”，确保你的 Session 在使用完后自动关闭。

小任务: 使用 with 语句，在 Session 中执行多个数据库操作。

14. 使用 `scoped_session` 管理 Session

痛点: 在多线程或 web 应用中管理 Session 很复杂。

方案: 使用 scoped_session 管理 Session。

from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.orm import sessionmaker, scoped_session
from sqlalchemy.ext.declarative import declarative_base
import threading

engine = create_engine('sqlite:///:memory:')
Base = declarative_base()

classUser(Base):
    __tablename__ = 'users'
    id = Column(Integer, primary_key=True)
    name = Column(String)

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

Session = scoped_session(Session)

defadd_user(name):
    session = Session()
try:
        user = User(name=name)
        session.add(user)
        session.commit()
        print(f"Added user: {name} in thread {threading.get_ident()}")

finally:
        Session.remove()

threads = []
for i in range(2):
    t = threading.Thread(target=add_user, args=[f'User{i}'])
    threads.append(t)
    t.start()

for t in threads:
   t.join()

with Session() as session:
    users = session.query(User).all()
for user in users:
       print(f"Retrieved user {user.name}")


# 输出 (顺序可能会不同)
# Added user: User0 in thread 140141813962496
# Added user: User1 in thread 140141805560064
# Retrieved user User0
# Retrieved user User1

解释:scoped_session 可以让每个线程都拥有独立的 Session 对象，避免多线程问题；使用 Session.remove() 来清理线程内的 Session。类比:scoped_session 就像一个 “线程隔离器”，确保每个线程拥有独立的 Session，不会互相干扰。

小任务: 使用 scoped_session 创建一个简单的 Web 应用，并测试 Session 的管理是否正确。

15. 使用 `lazy="joined"` 优化关联查询

痛点: 默认的延迟加载 (lazy) 模式，导致 N+1 查询问题。

方案: 使用 lazy="joined" 模式，实现立即加载关联数据。

from sqlalchemy import create_engine, Column, Integer, String, ForeignKey
from sqlalchemy.orm import relationship, sessionmaker
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import event

engine = create_engine('sqlite:///:memory:', echo=True) # echo=True 可以显示sql语句
Base = declarative_base()

classAuthor(Base):
    __tablename__ = 'authors'
    id = Column(Integer, primary_key=True)
    name = Column(String)
    books = relationship("Book", backref="author", lazy="joined")

classBook(Base):
    __tablename__ = 'books'
    id = Column(Integer, primary_key=True)
    title = Column(String)
    author_id = Column(Integer, ForeignKey('authors.id'))

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

author = Author(name='Jane Austen')
book1 = Book(title='Pride and Prejudice', author=author)
book2 = Book(title='Sense and Sensibility', author=author)

session.add(author)
session.add(book1)
session.add(book2)
session.commit()

retrieved_author = session.query(Author).filter_by(name='Jane Austen').first()

print(f"\nAuthor: {retrieved_author.name}")
for book in retrieved_author.books:
    print(f"  Book: {book.title}")


session.close()

# 输出 (注意观察SQL语句，只有一条查询)
# SELECT authors.id AS authors_id, authors.name AS authors_name
# FROM authors
# WHERE authors.name = ?
#
# SELECT books.id AS books_id, books.title AS books_title, books.author_id AS books_author_id
# FROM books
# WHERE books.author_id = ?

# Author: Jane Austen
#   Book: Pride and Prejudice
#   Book: Sense and Sensibility

解释:lazy="joined" 表示在查询 Author 时，立即加载关联的 Book 对象; echo=True 可以打印 SQL 语句，方便观察查询次数。类比:lazy="joined" 就像“立即加载”，在查询的时候，把所有需要的数据一次性加载过来，避免多次查询。

小任务: 对比 lazy="select"（默认值）和 lazy="joined" 的查询次数，分析性能差异。

16. 使用 `lazy="subquery"` 优化复杂的关联查询

痛点:lazy="joined" 无法满足复杂的关联查询。

方案: 使用 lazy="subquery" 实现子查询加载关联数据。

from sqlalchemy import create_engine, Column, Integer, String, ForeignKey
from sqlalchemy.orm import relationship, sessionmaker
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import event

engine = create_engine('sqlite:///:memory:', echo=True)
Base = declarative_base()

classAuthor(Base):
    __tablename__ = 'authors'
    id = Column(Integer, primary_key=True)
    name = Column(String)
    books = relationship("Book", backref="author", lazy="subquery")

classBook(Base):
    __tablename__ = 'books'
    id = Column(Integer, primary_key=True)
    title = Column(String)
    author_id = Column(Integer, ForeignKey('authors.id'))

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

author = Author(name='Jane Austen')
book1 = Book(title='Pride and Prejudice', author=author)
book2 = Book(title='Sense and Sensibility', author=author)

session.add(author)
session.add(book1)
session.add(book2)
session.commit()

retrieved_author = session.query(Author).filter_by(name='Jane Austen').first()

print(f"\nAuthor: {retrieved_author.name}")
for book in retrieved_author.books:
    print(f"  Book: {book.title}")


session.close()

# 输出 (注意观察SQL语句, 会使用子查询)
# SELECT authors.id AS authors_id, authors.name AS authors_name
# FROM authors
# WHERE authors.name = ?
#
# SELECT books.id AS books_id, books.title AS books_title, books.author_id AS books_author_id
# FROM books
# WHERE books.author_id IN (SELECT authors.id
# FROM authors
# WHERE authors.name = ?)
# Author: Jane Austen
#   Book: Pride and Prejudice
#   Book: Sense and Sensibility

解释:lazy="subquery" 表示使用子查询的方式加载关联数据，可以优化复杂的关联查询。类比:lazy="subquery" 就像一个 “子查询加载器”，在需要的时候，用子查询加载关联数据，可以更灵活。

小任务: 对比 lazy="joined" 和 lazy="subquery" 的 SQL 查询语句，分析其区别。

17. 使用 `lazy="dynamic"` 动态加载关联数据

痛点: 有些关联数据不需要每次都加载。

方案: 使用 lazy="dynamic" 延迟加载关联数据，并返回一个可查询的集合。

from sqlalchemy import create_engine, Column, Integer, String, ForeignKey
from sqlalchemy.orm import relationship, sessionmaker
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import event

engine = create_engine('sqlite:///:memory:', echo=True)
Base = declarative_base()

classAuthor(Base):
    __tablename__ = 'authors'
    id = Column(Integer, primary_key=True)
    name = Column(String)
    books = relationship("Book", backref="author", lazy="dynamic")

classBook(Base):
    __tablename__ = 'books'
    id = Column(Integer, primary_key=True)
    title = Column(String)
    author_id = Column(Integer, ForeignKey('authors.id'))

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

author = Author(name='Jane Austen')
book1 = Book(title='Pride and Prejudice', author=author)
book2 = Book(title='Sense and Sensibility', author=author)

session.add(author)
session.add(book1)
session.add(book2)
session.commit()

retrieved_author = session.query(Author).filter_by(name='Jane Austen').first()

print(f"\nAuthor: {retrieved_author.name}")
# 注意：这里 retrieved_author.books 返回的是一个 Query 对象，需要 all() 来获取数据
for book in retrieved_author.books.all():
    print(f"  Book: {book.title}")

session.close()

# 输出 (观察sql语句，只有在遍历 books 时才会查询)
# SELECT authors.id AS authors_id, authors.name AS authors_name
# FROM authors
# WHERE authors.name = ?
#
# SELECT books.id AS books_id, books.title AS books_title, books.author_id AS books_author_id
# FROM books
# WHERE books.author_id = ?
#
# Author: Jane Austen
#   Book: Pride and Prejudice
#   Book: Sense and Sensibility

解释:lazy="dynamic" 表示返回一个 Query 对象，你可以按需查询关联数据，适合关联数据量大的场景。类比:lazy="dynamic" 就像一个 “按需加载器”，只有在需要的时候才去加载关联数据。

小任务: 使用 lazy="dynamic"，并在获取关联数据时添加过滤条件。

18. 使用 `with_entities()` 指定查询返回列

痛点: 只想查询特定列，而不是所有列。

方案: 使用 with_entities() 指定查询返回列。

from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.orm import sessionmaker
from sqlalchemy.ext.declarative import declarative_base

engine = create_engine('sqlite:///:memory:')
Base = declarative_base()

classProduct(Base):
    __tablename__ = 'products'
    id = Column(Integer, primary_key=True)
    name = Column(String)
    price = Column(Integer)

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

session.add(Product(name='Laptop', price=1200))
session.commit()

# 使用 with_entities 只查询 name 和 price
product = session.query(Product).with_entities(Product.name, Product.price).first()
print(f"Product name: {product.name}, price: {product.price}")

session.close()

# 输出
# Product name: Laptop, price: 1200

解释:with_entities() 指定查询返回的列，提高查询效率。类比:with_entities() 就像一个 “列选择器”，你可以选择需要查询的列，避免返回不必要的数据。

小任务: 使用 with_entities() 查询多个表的特定列，并进行连接查询。

19. 使用 `defer()` 延迟加载列

痛点: 有些列不需要每次都加载。

方案: 使用 defer() 延迟加载指定的列。

from sqlalchemy import create_engine, Column, Integer, String, Text
from sqlalchemy.orm import sessionmaker
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import event

engine = create_engine('sqlite:///:memory:', echo=True) # echo=True 可以打印sql

Base = declarative_base()

classArticle(Base):
    __tablename__ = 'articles'
    id = Column(Integer, primary_key=True)
    title = Column(String)
    content = Column(Text)

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

article = Article(title='Test Article', content='This is a long content for the article.')
session.add(article)
session.commit()


# 延迟加载 content 列
retrieved_article = session.query(Article).options(defer(Article.content)).first()

print(f"Article title: {retrieved_article.title}") # 这时 content 没有被加载
print(f"Article content: {retrieved_article.content}") #这时 content 被加载

session.close()

# 输出 (观察SQL语句，发现content 列被分两次查询)
# SELECT articles.id AS articles_id, articles.title AS articles_title
# FROM articles
# LIMIT ? OFFSET ?
#
# Article title: Test Article
#
# SELECT articles.content AS articles_content
# FROM articles
# WHERE articles.id = ?
#
# Article content: This is a long content for the article.

解释:defer() 延迟加载指定的列，直到真正访问该属性时才查询，提高查询效率。类比:defer() 就像一个“延迟加载器”，只有在真正需要的时候才加载指定的列。

小任务: 使用 defer() 延迟加载文章内容，并对比加载时间和内存占用情况。

20. 使用 `load_only()` 只加载指定的列

痛点: 只想加载特定列，而不是所有列。

方案: 使用 load_only() 只加载指定的列。

from sqlalchemy import create_engine, Column, Integer, String, Text
from sqlalchemy.orm import sessionmaker
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import event

engine = create_engine('sqlite:///:memory:', echo=True)

Base = declarative_base()

classArticle(Base):
    __tablename__ = 'articles'
    id = Column(Integer, primary_key=True)
    title = Column(String)
    content = Column(Text)

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

article = Article(title='Test Article', content='This is a long content for the article.')
session.add(article)
session.commit()

#  只加载 title 列
retrieved_article = session.query(Article).options(load_only(Article.title)).first()

print(f"Article title: {retrieved_article.title}")

#print(f"Article content: {retrieved_article.content}") #  这里访问 content 会报错

session.close()

# 输出 (观察SQL语句，只查询title列)
# SELECT articles.id AS articles_id, articles.title AS articles_title
# FROM articles
# LIMIT ? OFFSET ?
#
# Article title: Test Article

解释:load_only() 只加载指定的列，其他列不会加载，提高查询效率。类比:load_only() 就像一个 “列过滤器”，让你只加载需要的列，避免加载不必要的列。

小任务: 使用 load_only() 只加载文章的标题，并对比加载时间和内存占用情况。

21. 使用 `with_polymorphic` 进行多态查询

痛点: 需要处理继承关系的查询。

方案: 使用 with_polymorphic() 进行多态查询。

from sqlalchemy import create_engine, Column, Integer, String, ForeignKey
from sqlalchemy.orm import sessionmaker, relationship, with_polymorphic
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import event

engine = create_engine('sqlite:///:memory:', echo=True) # echo=True 打印sql
Base = declarative_base()

classEmployee(Base):
    __tablename__ = 'employees'
    id = Column(Integer, primary_key=True)
    type = Column(String)  # 用于标识类型
    name = Column(String)
    __mapper_args__ = {
'polymorphic_identity':'employee',
'polymorphic_on': type
    }

classManager(Employee):
    __tablename__ = 'managers'
    id = Column(Integer, ForeignKey('employees.id'), primary_key=True)
    department = Column(String)
    __mapper_args__ = {
'polymorphic_identity':'manager',
    }

classEngineer(Employee):
    __tablename__ = 'engineers'
    id = Column(Integer, ForeignKey('employees.id'), primary_key=True)
    language = Column(String)
    __mapper_args__ = {
'polymorphic_identity':'engineer'
    }


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

manager = Manager(name='Alice', department='IT')
engineer = Engineer(name='Bob', language='Python')

session.add(manager)
session.add(engineer)
session.commit()

#使用 with_polymorphic 进行多态查询
employee_poly = with_polymorphic(Employee, [Manager, Engineer])
employees = session.query(employee_poly).all()
for emp in employees:
    print(f"Employee: {emp.name}, Type:{emp.type}", end ="")
if isinstance(emp, Manager):
      print(f", Department: {emp.department}")
elif isinstance(emp, Engineer):
      print(f", Language: {emp.language}")
session.close()


# 输出(注意查询语句，只会查询一个表)
# SELECT employees.id AS employees_id, employees.type AS employees_type, employees.name AS employees_name, managers.department AS managers_department
# FROM employees LEFT OUTER JOIN managers ON employees.id = managers.id
# WHERE employees.type IN (?, ?)
#
# SELECT employees.id AS employees_id, employees.type AS employees_type, employees.name AS employees_name, engineers.language AS engineers_language
# FROM employees LEFT OUTER JOIN engineers ON employees.id = engineers.id
# WHERE employees.type IN (?, ?)

# Employee: Alice, Type:manager, Department: IT
# Employee: Bob, Type:engineer, Language: Python

解释:with_polymorphic() 处理继承关系的查询，减少查询次数。类比:with_polymorphic() 就像一个 “多态查询器”，可以统一查询不同类型的数据。

小任务: 使用 with_polymorphic()，查询所有员工的姓名，并根据类型打印不同的信息。

22. 使用 `session.bulk_insert_mappings()` 批量插入数据

痛点: 循环插入数据，效率低下。

方案: 使用 session.bulk_insert_mappings() 批量插入数据。

from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.orm import sessionmaker
from sqlalchemy.ext.declarative import declarative_base

engine = create_engine('sqlite:///:memory:')
Base = declarative_base()

classUser(Base):
    __tablename__ = 'users'
    id = Column(Integer, primary_key=True)
    name = Column(String)

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

users = [{'name': f'User {i}'} for i in range(1000)]
session.bulk_insert_mappings(User, users)
session.commit()

print(f"Inserted {len(users)} users")

session.close()

# 输出
# Inserted 1000 users

解释:bulk_insert_mappings() 方法批量插入数据，减少数据库交互次数，提高插入效率。类比:bulk_insert_mappings() 就像一个“批量插入器”，一次性插入多条数据，避免多次插入操作。

小任务: 使用 bulk_insert_mappings() 批量插入 10000 条数据，并计算插入耗时。

23. 使用 `session.bulk_update_mappings()` 批量更新数据

痛点: 循环更新数据，效率低下。

方案: 使用 session.bulk_update_mappings() 批量更新数据。

from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.orm import sessionmaker
from sqlalchemy.ext.declarative import declarative_base

engine = create_engine('sqlite:///:memory:')
Base = declarative_base()

classProduct(Base):
    __tablename__ = 'products'
    id = Column(Integer, primary_key=True)
    name = Column(String)
    price = Column(Integer)

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

products = [{'id': i, 'name': f'Product {i}', 'price': 100} for i in range(1, 1001)]
session.bulk_insert_mappings(Product, products)
session.commit()

#批量更新
updates = [{'id': i, 'price': 200} for i in range(1,1001) if i % 2 == 0]
session.bulk_update_mappings(Product, updates)
session.commit()


retrieved_products = session.query(Product).filter(Product.id % 2 == 0).all()
for prod in retrieved_products:
    print(f"Product ID: {prod.id}, Name:{prod.name}, Updated price: {prod.price}")
session.close()
#输出 (只会显示id为偶数的商品)
# Product ID: 2, Name:Product 2, Updated price: 200
# Product ID: 4, Name:Product 4, Updated price: 200
# Product ID: 6, Name:Product 6, Updated price: 200
# ...

解释:bulk_update_mappings() 方法批量更新数据，减少数据库交互次数，提高更新效率。类比:bulk_update_mappings() 就像一个“批量更新器”，一次性更新多条数据，避免多次更新操作。

小任务: 使用 bulk_update_mappings() 批量更新 10000 条数据，并计算更新耗时。

24. 使用原生SQL语句执行复杂查询

痛点: ORM 无法满足复杂的 SQL 查询需求。

方案: 使用 session.execute() 执行原生 SQL 语句。

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

engine = create_engine('sqlite:///:memory:')
Base = declarative_base()

classUser(Base):
    __tablename__ = 'users'
    id = Column(Integer, primary_key=True)
    name = Column(String)
    age = Column(Integer)

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

session.add_all([
  User(name='Alice', age=25),
  User(name='Bob', age = 30),
  User(name='Charlie', age = 22)
])
session.commit()

# 使用 text 执行原生 SQL 查询
sql_query = text("SELECT name, age FROM users WHERE age > :age_param ORDER BY age DESC")
result = session.execute(sql_query, {"age_param": 23}).all() # 注意参数要使用字典
print("Users with age > 23:")
for row in result:
    print(f"- Name: {row.name}, Age: {row.age}")
session.close()

# 输出
# Users with age > 23:
# - Name: Bob, Age: 30
# - Name: Alice, Age: 25

解释: 使用 session.execute() 执行原生 SQL 语句，可以满足复杂的 SQL 查询需求; text()函数创建SQL表达式，参数使用字典传递。类比:session.execute() 就像一个“SQL直通车”，可以直接执行 SQL 语句，灵活性非常高。

小任务: 使用 session.execute() 执行带有 JOIN 操作的复杂 SQL 查询。

25. 使用 `hybrid_property` 创建混合属性

痛点: 需要根据对象属性计算出新的值

方案: 使用 hybrid_property 创建混合属性，可以在 Python 层和 SQL 层使用

from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.orm import

开篇：告别低效ORM，拥抱高效SQLAlchemy

1. 基础篇：理解 Session 的生命周期

2. 使用 Declarative Base 定义模型

3. 使用 Relationships 处理关联关系

4. 使用 Query 的各种过滤条件

5. 使用 limit 和 offset 分页查询

6. 使用 order_by 进行排序

7. 使用 group_by 和 having 进行分组统计

8. 使用 join 进行多表连接查询

9. 使用 Subquery 进行子查询

10. 使用 exists() 进行存在性判断

11. 使用 update() 方法进行数据更新

12. 使用 delete() 方法删除数据

13. 使用 with 语句管理 Session 上下文

14. 使用 scoped_session 管理 Session

15. 使用 lazy="joined" 优化关联查询

16. 使用 lazy="subquery" 优化复杂的关联查询

17. 使用 lazy="dynamic" 动态加载关联数据

18. 使用 with_entities() 指定查询返回列

19. 使用 defer() 延迟加载列

20. 使用 load_only() 只加载指定的列

21. 使用 with_polymorphic 进行多态查询

22. 使用 session.bulk_insert_mappings() 批量插入数据

23. 使用 session.bulk_update_mappings() 批量更新数据