Python中SQLAlchemy ORM应用实践
环境配置
安装SQLAlchemy核心库及数据库驱动：
# 基础安装
pip install sqlalchemy
# 可选数据库驱动
pip install psycopg2-binary # PostgreSQL
pip install pymysql # MySQL核心组件
- 引擎(Engine)：数据库连接枢纽
- 会话(Session)：持久化操作管理器
- 数据模型(Model)：表结构映射类
- 查询(Query)：数据库检索构造器
数据库连接
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
# 配置数据库连接
db_engine = create_engine('postgresql://user:pass@localhost:5432/mydb')
# 创建会话工厂
SessionFactory = sessionmaker(bind=db_engine, autoflush=False)
# 实例化会话
db_session = SessionFactory()模型定义
from sqlalchemy import Column, Integer, String, ForeignKey
from sqlalchemy.orm import relationship, declarative_base
ModelBase = declarative_base()
class Customer(ModelBase):
__tablename__ = 'customers'
customer_id = Column(Integer, primary_key=True)
full_name = Column(String(50))
contact_email = Column(String(100), unique=True)
orders = relationship("PurchaseOrder", back_populates="buyer")
class PurchaseOrder(ModelBase):
__tablename__ = 'orders'
order_id = Column(Integer, primary_key=True)
product_name = Column(String(100))
customer_ref = Column(Integer, ForeignKey('customers.customer_id'))
buyer = relationship("Customer", back_populates="orders")
# 初始化表结构
ModelBase.metadata.create_all(bind=db_engine)数据操作
创建记录
new_customer = Customer(full_name="李明", contact_email="liming@example.com")
db_session.add(new_customer)
db_session.commit()查询记录
# 条件查询
customers = db_session.query(Customer).filter(
Customer.full_name.like("李%")
).all()更新记录
customer = db_session.query(Customer).get(101)
customer.contact_email = "new_email@example.com"
db_session.commit()删除记录
db_session.query(Customer).filter_by(customer_id=102).delete()
db_session.commit()高级查询
from sqlalchemy import func
# 聚合查询
order_counts = db_session.query(
Customer.full_name,
func.count(PurchaseOrder.order_id)
).join(PurchaseOrder).group_by(Customer.customer_id).all()关系管理
# 添加关联对象
customer = Customer(full_name="张华")
new_order = PurchaseOrder(product_name="笔记本电脑", buyer=customer)
db_session.add(new_order)
db_session.commit()
# 遍历关系
for order in customer.orders:
print(f"订单: {order.product_name}")事务控制
try:
db_session.begin_nested()
# 执行数据库操作
db_session.commit()
except Exception:
db_session.rollback()优化实践
- 使用上下文管理器管理会话生命周期
- 批量操作代替逐条提交
- 预加载关联数据避免N+1查询
from contextlib import contextmanager
@contextmanager
def session_scope():
session = SessionFactory()
try:
yield session
session.commit()
except:
session.rollback()
raise
finally:
session.close()