Python SQLAlchemy ORM 数据库操作指南

创建数据库表

# 创建所有表
Base.metadata.create_all(bind=engine)

# 删除所有表
# Base.metadata.drop_all(bind=engine)

基本 CRUD 操作

创建数据

# 创建新用户
new_user = User(name="张三", email="[email protected]")
session.add(new_user)
session.commit()

# 批量创建
session.add_all([
    User(name="李四", email="[email protected]"),
    User(name="王五", email="[email protected]")
])
session.commit()

读取数据

# 获取所有用户
users = session.query(User).all()

# 获取第一个用户
first_user = session.query(User).first()

# 根据 ID 获取用户
user = session.query(User).get(1)

更新数据

# 查询并更新
user = session.query(User).get(1)
user.name = "张三四"
session.commit()

# 批量更新
session.query(User).filter(User.name.like("张%")).update({"name": "张氏"}, synchronize_session=False)
session.commit()

删除数据

# 查询并删除
user = session.query(User).get(1)
session.delete(user)
session.commit()

# 批量删除
session.query(User).filter(User.name == "李四").delete(synchronize_session=False)
session.commit()

查询数据

基本查询

# 获取所有记录
users = session.query(User).all()

# 获取特定字段
names = session.query(User.name).all()

# 排序
users = session.query(User).order_by(User.name.desc()).all()

# 限制结果数量
users = session.query(User).limit(10).all()

# 偏移量
users = session.query(User).offset(5).limit(10).all()

过滤查询

from sqlalchemy import or_

# 等值过滤
user = session.query(User).filter(User.name == "张三").first()

# 模糊查询
users = session.query(User).filter(User.name.like("张%")).all()

# IN 查询
users = session.query(User).filter(User.name.in_(["张三", "李四"])).all()

# 多条件查询
users = session.query(User).filter(
    User.name == "张三",
    User.email.like("%@example.com")
).all()

# 或条件
users = session.query(User).filter(
    or_(User.name == "张三", User.name == "李四")
).all()

# 不等于
users = session.query(User).filter(User.name != "张三").all()

聚合查询

from sqlalchemy import func

# 计数
count = session.query(User).count()

# 分组计数
user_post_count = session.query(
    User.name,
    func.count(Post.id)
).join(Post).group_by(User.name).all()

# 求和、平均值等
avg_id = session.query(func.avg(User.id)).scalar()

连接查询

# 内连接
results = session.query(User, Post).join(Post).filter(Post.title.like("%Python%")).all()

# 左外连接
results = session.query(User, Post).outerjoin(Post).all()

# 指定连接条件
results = session.query(User, Post).join(Post, User.id == Post.author_id).all()

关系操作

# 创建带关系的对象
user = User(name="赵六", email="[email protected]")
post = Post(title="我的第一篇博客", content="Hello World!", author=user)
session.add(post)
session.commit()

# 通过关系访问
print(f"文章 '{post.title}' 的作者是 {post.author.name}")
print(f"用户 {user.name} 的所有文章:")
for p in user.posts:
    print(f" - {p.title}")

# 多对多关系操作
python_tag = Tag(name="Python")
sqlalchemy_tag = Tag(name="SQLAlchemy")
post.tags.append(python_tag)
post.tags.append(sqlalchemy_tag)
session.commit()
print(f"文章 '{post.title}' 的标签:")
for tag in post.tags:
    print(f" - {tag.name}")

事务管理

# 自动提交事务
try:
    user = User(name="测试用户", email="[email protected]")
    session.add(user)
    session.commit()
except Exception as e:
    session.rollback()
    print(f"发生错误：{e}")

# 使用事务上下文管理器
from sqlalchemy.orm import Session

def create_user(session: Session, name: str, email: str):
    try:
        user = User(name=name, email=email)
        session.add(user)
        session.commit()
        return user
    except:
        session.rollback()
        raise

# 嵌套事务
with session.begin_nested():
    user = User(name="事务用户", email="[email protected]")
    session.add(user)
    # 保存点
    savepoint = session.begin_nested()
    try:
        user = User(name="保存点用户", email="[email protected]")
        session.add(user)
        savepoint.commit()
    except:
        savepoint.rollback()

最佳实践

1. 会话管理：为每个请求创建新会话，请求结束后关闭
2. 异常处理：始终处理异常并适当回滚事务
3. 延迟加载：注意 N+1 查询问题，使用 eager loading 优化
4. 连接池：合理配置连接池大小和超时设置
5. 数据验证：在模型层或应用层验证数据完整性

# 使用上下文管理器管理会话
from contextlib import contextmanager

@contextmanager
def get_db():
    db = SessionLocal()
    try:
        yield db
        db.commit()
    except Exception:
        db.rollback()
        raise
    finally:
        db.close()

# 使用示例
with get_db() as db:
    user = User(name="上下文用户", email="[email protected]")
    db.add(user)

总结

SQLAlchemy ORM 提供了强大而灵活的数据库操作方式，通过本文的介绍，您应该能够：

1. 安装和配置 SQLAlchemy
2. 定义数据模型和关系
3. 执行基本的 CRUD 操作
4. 构建复杂查询
5. 管理数据库事务
6. 遵循最佳实践

SQLAlchemy 还有更多高级特性，如混合属性、事件监听、自定义查询等，值得进一步探索学习。