Python SQLAlchemy ORM 数据库操作指南
SQLAlchemy 简介
SQLAlchemy 是 Python 生态中备受推崇的对象关系映射（ORM）框架，它为开发者提供了优雅且高效的数据库交互方式。本指南将深入探讨如何利用 SQLAlchemy ORM 进行各种数据库操作。
目录
- 环境准备与安装
- 核心架构组件
- 建立数据库连接
- 数据模型设计
- 数据库表结构生成
- 基础数据操作
- 高级查询技术
- 关联关系处理
- 事务控制机制
- 性能优化建议
环境准备与安装
首先，我们需要安装 SQLAlchemy 包：
pip install sqlalchemy针对不同的数据库系统，还需要安装相应的数据库驱动：
# PostgreSQL 数据库
pip install psycopg2-binary
# MySQL 数据库
pip install mysql-connector-python
# SQLite 数据库（Python 标准库已包含，无需额外安装）
核心架构组件
- 引擎（Engine）：负责与数据库建立连接并处理通信的核心组件
- 会话（Session）：管理数据库操作的生命周期，包括持久化、查询等
- 模型（Model）：定义数据结构的类，映射到数据库表
- 查询（Query）：构建和执行数据库查询的对象接口
建立数据库连接
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
# 创建数据库连接引擎
# SQLite 示例
db_engine = create_engine('sqlite:///company.db', echo=True)
# PostgreSQL 示例
# db_engine = create_engine('postgresql://user:password@localhost:5432/enterprise_db')
# MySQL 示例
# db_engine = create_engine('mysql+mysqlconnector://user:password@localhost:3306/enterprise_db')
# 创建会话工厂
SessionFactory = sessionmaker(autocommit=False, autoflush=False, bind=db_engine)
# 创建会话实例
current_session = SessionFactory()
数据模型设计
from sqlalchemy import Column, Integer, String, ForeignKey, DateTime, Boolean
from sqlalchemy.orm import relationship, declarative_base
from datetime import datetime
# 创建基类
ModelBase = declarative_base()
class Employee(ModelBase):
__tablename__ = 'employees'
id = Column(Integer, primary_key=True, index=True)
name = Column(String(50), nullable=False)
email = Column(String(100), unique=True, index=True)
position = Column(String(50))
hire_date = Column(DateTime, default=datetime.utcnow)
is_active = Column(Boolean, default=True)
# 定义一对多关系
projects = relationship("Project", back_populates="team_lead")
assignments = relationship("Assignment", back_populates="employee")
class Project(ModelBase):
__tablename__ = 'projects'
id = Column(Integer, primary_key=True, index=True)
name = Column(String(100), nullable=False)
description = Column(String(500))
start_date = Column(DateTime, nullable=False)
end_date = Column(DateTime)
status = Column(String(20), default="进行中")
team_lead_id = Column(Integer, ForeignKey('employees.id'))
# 定义多对一关系
team_lead = relationship("Employee", back_populates="projects")
# 定义多对多关系（通过关联表）
team_members = relationship("Employee", secondary="project_assignments", back_populates="projects")
class Assignment(ModelBase):
__tablename__ = 'assignments'
id = Column(Integer, primary_key=True, index=True)
project_id = Column(Integer, ForeignKey('projects.id'), nullable=False)
employee_id = Column(Integer, ForeignKey('employees.id'), nullable=False)
role = Column(String(50))
hours_allocated = Column(Integer)
start_date = Column(DateTime, default=datetime.utcnow)
# 定义关系
project = relationship("Project")
employee = relationship("Employee", back_populates="assignments")
# 关联表（用于多对多关系）
class ProjectAssignment(ModelBase):
__tablename__ = 'project_assignments'
project_id = Column(Integer, ForeignKey('projects.id'), primary_key=True)
employee_id = Column(Integer, ForeignKey('employees.id'), primary_key=True)
assignment_date = Column(DateTime, default=datetime.utcnow)
role = Column(String(50))
数据库表结构生成
# 创建所有定义的表
ModelBase.metadata.create_all(bind=db_engine)
# 删除所有表（谨慎使用）
# ModelBase.metadata.drop_all(bind=db_engine)
基础数据操作
创建数据
# 创建新员工
new_employee = Employee(name="李明", email="liming@company.com", position="软件工程师")
current_session.add(new_employee)
current_session.commit()
# 批量创建
current_session.add_all([
Employee(name="王芳", email="wangfang@company.com", position="产品经理"),
Employee(name="张伟", email="zhangwei@company.com", position="UI设计师")
])
current_session.commit()
读取数据
# 获取所有员工
all_employees = current_session.query(Employee).all()
# 获取第一个员工
first_employee = current_session.query(Employee).first()
# 根据ID获取员工
employee = current_session.query(Employee).get(1)
更新数据
# 查询并更新
employee = current_session.query(Employee).get(1)
employee.position = "高级软件工程师"
current_session.commit()
# 批量更新
current_session.query(Employee).filter(Employee.name.like("张%")).update(
{"position": "高级设计师"}, synchronize_session=False
)
current_session.commit()
删除数据
# 查询并删除
employee = current_session.query(Employee).get(1)
current_session.delete(employee)
current_session.commit()
# 批量删除
current_session.query(Employee).filter(Employee.name == "王芳").delete(synchronize_session=False)
current_session.commit()
高级查询技术
基本查询
# 获取所有记录
employees = current_session.query(Employee).all()
# 获取特定字段
names = current_session.query(Employee.name).all()
# 排序
employees = current_session.query(Employee).order_by(Employee.name.desc()).all()
# 限制结果数量
employees = current_session.query(Employee).limit(10).all()
# 分页查询
page = 2
per_page = 5
employees = current_session.query(Employee).offset((page-1)*per_page).limit(per_page).all()
过滤查询
from sqlalchemy import or_, and_
# 等值过滤
employee = current_session.query(Employee).filter(Employee.name == "李明").first()
# 模糊查询
employees = current_session.query(Employee).filter(Employee.name.like("李%")).all()
# IN查询
employees = current_session.query(Employee).filter(Employee.name.in_(["李明", "王芳"])).all()
# 多条件查询
employees = current_session.query(Employee).filter(
and_(
Employee.name == "李明",
Employee.position.like("%工程师")
)
).all()
# 或条件
employees = current_session.query(Employee).filter(
or_(
Employee.name == "李明",
Employee.name == "王芳"
)
).all()
# 不等于
employees = current_session.query(Employee).filter(Employee.name != "李明").all()
聚合查询
from sqlalchemy import func
# 计数
employee_count = current_session.query(Employee).count()
# 分组计数
position_counts = current_session.query(
Employee.position,
func.count(Employee.id)
).group_by(Employee.position).all()
# 求和、平均值等
avg_hire_days = current_session.query(func.avg(func.datediff('now', Employee.hire_date))).scalar()
连接查询
# 内连接
project_assignments = current_session.query(Employee, Project).join(
Project, Employee.id == Project.team_lead_id
).filter(Project.status == "进行中").all()
# 左外连接
all_projects = current_session.query(Project).outerjoin(
Assignment, Project.id == Assignment.project_id
).all()
# 多表连接
employee_projects = current_session.query(Employee, Project, Assignment).join(
Project, Employee.id == Project.team_lead_id
).join(
Assignment, Project.id == Assignment.project_id
).filter(Employee.position == "软件工程师").all()
关联关系处理
# 创建带关系的对象
new_employee = Employee(name="陈静", email="chenjing@company.com", position="项目经理")
new_project = Project(
name="客户管理系统",
description="开发新的客户关系管理平台",
start_date=datetime.utcnow(),
team_lead=new_employee
)
current_session.add(new_project)
current_session.commit()
# 通过关系访问
print(f"项目 '{new_project.name}' 的负责人是 {new_project.team_lead.name}")
print(f"员工 {new_employee.name} 负责的项目:")
for p in new_employee.projects:
print(f" - {p.name}")
# 多对多关系操作
team_member1 = Employee(name="刘强", email="liuqiang@company.com", position="前端开发")
team_member2 = Employee(name="赵敏", email="zhaomin@company.com", position="后端开发")
new_project.team_members.append(team_member1)
new_project.team_members.append(team_member2)
current_session.commit()
print(f"项目 '{new_project.name}' 的团队成员:")
for member in new_project.team_members:
print(f" - {member.name} ({member.position})")
事务控制机制
# 自动提交事务
try:
employee = Employee(name="测试员工", email="test@company.com", position="测试工程师")
current_session.add(employee)
current_session.commit()
except Exception as e:
current_session.rollback()
print(f"发生错误: {e}")
# 使用事务上下文管理器
from sqlalchemy.orm import Session
def add_employee(session: Session, name: str, email: str, position: str):
try:
employee = Employee(name=name, email=email, position=position)
session.add(employee)
session.commit()
return employee
except:
session.rollback()
raise
# 嵌套事务
with current_session.begin_nested():
employee = Employee(name="事务员工", email="transaction@company.com", position="事务工程师")
current_session.add(employee)
# 保存点
savepoint = current_session.begin_nested()
try:
employee = Employee(name="保存点员工", email="savepoint@company.com", position="保存点工程师")
current_session.add(employee)
savepoint.commit()
except:
savepoint.rollback()
性能优化建议
- 会话管理：为每个请求创建新会话，请求结束后关闭
- 异常处理：始终处理异常并适当回滚事务
- 延迟加载：注意N+1查询问题，使用预加载（eager loading）优化
- 连接池配置：合理配置连接池大小和超时设置
- 数据验证：在模型层或应用层验证数据完整性
# 使用上下文管理器管理会话
from contextlib import contextmanager
@contextmanager
def get_database_session():
db_session = SessionFactory()
try:
yield db_session
db_session.commit()
except Exception:
db_session.rollback()
raise
finally:
db_session.close()
# 使用示例
with get_database_session() as session:
employee = Employee(name="上下文员工", email="context@company.com", position="上下文工程师")
session.add(employee)
总结
SQLAlchemy ORM 提供了强大而灵活的数据库操作方式，通过本指南的介绍，您应该能够：
- 安装和配置 SQLAlchemy
- 设计数据模型和关系
- 执行基本的 CRUD 操作
- 构建复杂查询
- 管理数据库事务
- 遵循最佳实践优化性能
SQLAlchemy 还有更多高级特性，如混合属性、事件监听、自定义查询等，值得进一步探索学习。
