如何在Python SQLAlchemy函数中实现类似FastAPI的DB会话管理？

你完全可以复用FastAPI中get_db的连接管理逻辑，不需要重复写手动开闭的代码，这里提供两种实用方案：

方案一：将get_db转为上下文管理器（推荐） #

利用Python的contextlib.contextmanager装饰器，把现有的get_db生成器函数转换成支持with语句的上下文管理器，这样在普通函数里就能像FastAPI内部一样自动管理连接的开闭：

首先修改get_db函数（如果不想改动原函数，可以封装一个新的）：

from contextlib import contextmanager
from sqlalchemy.orm import Session

# 假设你的SessionLocal已经定义好
# SessionLocal = sessionmaker(...)

@contextmanager
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

然后在do_something中使用：

def do_something():
    with get_db() as db:
        # 执行数据库操作
        person = db.query(models.Person).first()
        # 处理数据
        person.age += 1
        # 提交修改
        db.commit()
        # 查询关联数据
        cars = db.query(models.Car).filter_by(person_id=person.id).all()
        # 其他业务逻辑

这个方案完全复用了原有的连接管理逻辑，with块结束时会自动触发finally中的db.close()，和FastAPI的Depends行为一致。

方案二：直接调用生成器（不推荐，代码冗余） #

如果不想修改get_db，可以手动遍历生成器来触发连接的创建和关闭：

def do_something():
    db_generator = get_db()
    db = next(db_generator)
    try:
        # 数据库操作逻辑
        person = db.query(models.Person).first()
        db.add(models.Person(name="New Person"))
        db.commit()
    finally:
        # 触发生成器的finally块关闭连接
        try:
            next(db_generator)
        except StopIteration:
            pass

这种写法需要手动处理生成器的迭代，代码不够简洁，不如上下文管理器方案直观。

注意点 #

• 你原来写的do_something里用了yield db是错误的，这会把它变成一个生成器函数，而不是执行业务逻辑的普通函数，需要去掉yield。
• 两种方案都能保证数据库连接在操作完成后自动关闭，避免连接泄漏。

内容的提问来源于stack exchange，提问作者gilvictor