好的，各位观众老爷们，欢迎来到今天的“SQLAlchemy ORM 高级查询：复杂联表、子查询与事务管理”讲座现场！我是你们的老朋友，今天要跟大家聊聊SQLAlchemy ORM里那些高级操作，保证让你的数据库查询能力提升一个档次！
咱们今天的主题是：
- 复杂联表查询： 告别单表查询的寂寞，体验多表联动的快乐！
- 子查询： 查询中的查询，让你的SQL语句更加灵活多变！
- 事务管理： 确保数据的一致性和完整性，不再担心数据丢失或出错！
准备好了吗？让我们开始吧！
一、复杂联表查询：告别单身，拥抱多表关系！
在实际项目中，数据往往不是孤立存在的，而是通过各种关系连接在一起的。比如，一个用户可以有多个订单，一个订单可以包含多个商品。这时候，我们就需要用到联表查询来获取我们需要的数据。
SQLAlchemy ORM 提供了多种联表查询的方式，包括 join()、outerjoin()、leftjoin()、rightjoin()等等。 它们分别对应 SQL 中的 INNER JOIN、FULL OUTER JOIN、LEFT OUTER JOIN 和 RIGHT OUTER JOIN。
咱们先来定义几个模型：
from sqlalchemy import create_engine, Column, Integer, String, ForeignKey, Float
from sqlalchemy.orm import sessionmaker, relationship
from sqlalchemy.ext.declarative import declarative_base
Base = declarative_base()
class User(Base):
__tablename__ = 'users'
id = Column(Integer, primary_key=True)
name = Column(String)
email = Column(String)
orders = relationship("Order", back_populates="user") # 关联关系
def __repr__(self):
return f"<User(id={self.id}, name='{self.name}', email='{self.email}')>"
class Order(Base):
__tablename__ = 'orders'
id = Column(Integer, primary_key=True)
user_id = Column(Integer, ForeignKey('users.id'))
order_date = Column(String) # 简化处理，实际应使用 Date 或 DateTime 类型
user = relationship("User", back_populates="orders") # 反向关联关系
order_items = relationship("OrderItem", back_populates="order")
def __repr__(self):
return f"<Order(id={self.id}, user_id={self.user_id}, order_date='{self.order_date}')>"
class Product(Base):
__tablename__ = 'products'
id = Column(Integer, primary_key=True)
name = Column(String)
price = Column(Float)
order_items = relationship("OrderItem", back_populates="product")
def __repr__(self):
return f"<Product(id={self.id}, name='{self.name}', price={self.price})>"
class OrderItem(Base):
__tablename__ = 'order_items'
id = Column(Integer, primary_key=True)
order_id = Column(Integer, ForeignKey('orders.id'))
product_id = Column(Integer, ForeignKey('products.id'))
quantity = Column(Integer)
order = relationship("Order", back_populates="order_items")
product = relationship("Product", back_populates="order_items")
def __repr__(self):
return f"<OrderItem(id={self.id}, order_id={self.order_id}, product_id={self.product_id}, quantity={self.quantity})>"
# 创建引擎和会话
engine = create_engine('sqlite:///:memory:') # 使用内存数据库方便演示
Base.metadata.create_all(engine)
Session = sessionmaker(bind=engine)
session = Session()
# 插入一些测试数据
user1 = User(name='Alice', email='[email protected]')
user2 = User(name='Bob', email='[email protected]')
session.add_all([user1, user2])
session.commit()
order1 = Order(user_id=user1.id, order_date='2023-10-26')
order2 = Order(user_id=user1.id, order_date='2023-10-27')
order3 = Order(user_id=user2.id, order_date='2023-10-28')
session.add_all([order1, order2, order3])
session.commit()
product1 = Product(name='Laptop', price=1200.00)
product2 = Product(name='Mouse', price=25.00)
product3 = Product(name='Keyboard', price=75.00)
session.add_all([product1, product2, product3])
session.commit()
order_item1 = OrderItem(order_id=order1.id, product_id=product1.id, quantity=1)
order_item2 = OrderItem(order_id=order1.id, product_id=product2.id, quantity=2)
order_item3 = OrderItem(order_id=order2.id, product_id=product3.id, quantity=1)
order_item4 = OrderItem(order_id=order3.id, product_id=product1.id, quantity=1)
session.add_all([order_item1, order_item2, order_item3, order_item4])
session.commit()
好，模型和数据都准备好了，现在咱们来玩点刺激的！
1. join()：内连接，只返回匹配的行
假设我们要查询所有用户的订单信息，包括用户名和订单日期。
# 查询所有用户的订单信息，包括用户名和订单日期
results = session.query(User.name, Order.order_date).join(Order, User.id == Order.user_id).all()
for name, order_date in results:
print(f"User: {name}, Order Date: {order_date}")这个例子中，join(Order, User.id == Order.user_id) 表示将 User 表和 Order 表进行内连接，连接条件是 User.id == Order.user_id。 只有当 User 表和 Order 表中 user_id 匹配时，才会返回结果。
2. outerjoin() (或者 leftjoin()/rightjoin()): 外连接，返回所有行
外连接可以分为左外连接、右外连接和全外连接。
- 左外连接（
leftjoin()）: 返回左表的所有行，以及右表中与左表匹配的行。如果右表中没有与左表匹配的行，则右表的列显示为NULL。 - 右外连接（
rightjoin()）: 返回右表的所有行，以及左表中与右表匹配的行。如果左表中没有与右表匹配的行，则左表的列显示为NULL。 - 全外连接（
outerjoin()）: 返回左表和右表的所有行。如果左表中没有与右表匹配的行，则右表的列显示为NULL。如果右表中没有与左表匹配的行，则左表的列显示为NULL。
咱们以左外连接为例，查询所有用户的信息，包括他们的订单信息（如果用户有订单）。
# 查询所有用户的信息，包括他们的订单信息（如果用户有订单）
results = session.query(User.name, Order.order_date).outerjoin(Order, User.id == Order.user_id).all() #等价于 .leftjoin
for name, order_date in results:
print(f"User: {name}, Order Date: {order_date}") # order_date 可能为None在这个例子中，即使某个用户没有订单，也会返回该用户的姓名，但是 order_date 会显示为 None。
3. 多表连接：连接多个表，获取更丰富的信息
咱们可以连接多个表，获取更丰富的信息。 比如，查询所有订单的信息，包括用户名、订单日期和商品名称。
# 查询所有订单的信息，包括用户名、订单日期和商品名称
results = session.query(User.name, Order.order_date, Product.name).
join(Order, User.id == Order.user_id).
join(OrderItem, Order.id == OrderItem.order_id).
join(Product, Product.id == OrderItem.product_id).all()
for user_name, order_date, product_name in results:
print(f"User: {user_name}, Order Date: {order_date}, Product: {product_name}")这个例子中，我们连接了 User、Order、OrderItem 和 Product 四个表，获取了用户姓名、订单日期和商品名称。 是不是感觉数据一下子就丰富起来了？
二、子查询：查询中的查询，灵活多变！
子查询是指在一个查询语句中嵌套另一个查询语句。子查询可以出现在 SELECT、FROM、WHERE 和 HAVING 子句中。
SQLAlchemy ORM 提供了多种方式来使用子查询。
1. 在 WHERE 子句中使用子查询：
假设我们要查询购买了 "Laptop" 的用户的姓名。
# 查询购买了 "Laptop" 的用户的姓名
subquery = session.query(OrderItem.order_id).join(Product, Product.id == OrderItem.product_id).filter(Product.name == 'Laptop').subquery()
results = session.query(User.name).join(Order, User.id == Order.user_id).filter(Order.id.in_(subquery)).all()
for name, in results:
print(f"User who bought Laptop: {name}")在这个例子中，我们首先使用子查询 session.query(OrderItem.order_id).join(Product, Product.id == OrderItem.product_id).filter(Product.name == 'Laptop').subquery() 获取所有购买了 "Laptop" 的订单 ID，然后使用 filter(Order.id.in_(subquery)) 筛选出包含这些订单的用户。
2. 在 FROM 子句中使用子查询：
假设我们要查询每个用户购买的商品数量的总和。
# 查询每个用户购买的商品数量的总和
subquery = session.query(Order.user_id, func.count(OrderItem.id).label('total_items')).
join(OrderItem, Order.id == OrderItem.order_id).
group_by(Order.user_id).subquery()
results = session.query(User.name, subquery.c.total_items).
join(subquery, User.id == subquery.c.user_id).all()
for name, total_items in results:
print(f"User: {name}, Total Items: {total_items}")在这个例子中，我们首先使用子查询 session.query(Order.user_id, func.count(OrderItem.id).label('total_items')).join(OrderItem, Order.id == OrderItem.order_id).group_by(Order.user_id).subquery() 获取每个用户的订单数量，然后将这个子查询作为一个表，与 User 表进行连接，获取用户姓名和订单数量。 func.count 是 SQLAlchemy 提供的函数，用于进行聚合操作。
3. 使用 exists() 子查询
假设我们要查询至少有一个订单的用户的姓名。
from sqlalchemy import exists
# 查询至少有一个订单的用户的姓名
subquery = exists().where(Order.user_id == User.id)
results = session.query(User.name).filter(subquery).all()
for name, in results:
print(f"User with at least one order: {name}")
这个例子使用了 exists() 函数，它会检查子查询是否返回任何行。 如果返回任何行，则 exists() 返回 True，否则返回 False。
三、事务管理：确保数据的一致性和完整性！
事务是一系列数据库操作的集合，这些操作要么全部成功执行，要么全部失败回滚。事务可以确保数据的一致性和完整性，防止数据丢失或出错。
SQLAlchemy ORM 提供了简单的事务管理机制。
1. 开启事务：
当你创建一个 Session 对象时，实际上就开启了一个事务。
2. 提交事务：
使用 session.commit() 提交事务。 如果事务中的所有操作都成功执行，则会将更改保存到数据库中。
3. 回滚事务：
使用 session.rollback() 回滚事务。 如果事务中的任何操作失败，或者你需要撤销事务中的所有更改，则可以使用 rollback() 方法。
try:
# 开启事务（实际上 Session 创建时就已经开启了）
user = User(name='Charlie', email='[email protected]')
session.add(user)
session.commit() # 提交事务，保存数据
print("User added successfully.")
# 模拟一个错误
product = Product(name='Broken Product', price=-10.00) # 价格不能为负数
session.add(product)
session.commit() #这里会失败，因为价格校验失败
except Exception as e:
session.rollback() # 回滚事务，撤销所有更改
print(f"An error occurred: {e}. Transaction rolled back.")
finally:
session.close()在这个例子中，我们首先添加了一个用户，然后尝试添加一个价格为负数的商品。 由于价格不能为负数，因此添加商品的操作会失败，抛出异常。 在 except 块中，我们使用 session.rollback() 回滚事务，撤销了添加用户的操作。
总结：
今天我们学习了 SQLAlchemy ORM 的高级查询，包括复杂联表查询、子查询和事务管理。 掌握这些技巧，可以让你更加灵活地操作数据库，提高开发效率。
- 联表查询： 使用
join()、outerjoin()等方法连接多个表，获取更丰富的信息。 - 子查询： 在查询语句中嵌套另一个查询语句，实现更复杂的查询逻辑。
- 事务管理： 使用
session.commit()提交事务，使用session.rollback()回滚事务，确保数据的一致性和完整性。
希望今天的讲座对大家有所帮助！ 下次有机会再跟大家分享更多 SQLAlchemy ORM 的技巧！
友情提示：
- 实际项目中，数据库连接池的配置也很重要，可以提高性能。
- SQLAlchemy ORM 的功能非常强大，还有很多高级特性值得学习，比如事件监听、自定义类型等等。
- 要根据实际情况选择合适的查询方式，避免过度使用复杂的查询，影响性能。
最后，感谢大家的观看！ 记得点赞、收藏、转发哦！咱们下期再见！