如何基于SQLAlchemy构建基类复用引擎与会话连接

核心思路 #

把数据库连接、引擎初始化、会话管理、表反射这些通用逻辑封装到一个基类中，子类只需继承该基类，就能直接复用这些功能，无需重复编写连接代码。

重构后的代码实现 #

1. 创建通用数据库基类 DBBase #

from sqlalchemy import create_engine
from sqlalchemy.ext.automap import automap_base
from sqlalchemy.orm import scoped_session, sessionmaker
import config

class DBBase:
    # 全局初始化数据库引擎，仅创建一次
    engine = create_engine(config.DATABASE_URL)
    
    # 初始化automap基类，反射现有数据库的所有表
    Base = automap_base()
    Base.prepare(engine, reflect=True)
    
    # 配置线程安全的会话工厂，替代直接创建Session实例
    _session_factory = sessionmaker(bind=engine)
    session = scoped_session(_session_factory)

    @classmethod
    def get_session(cls):
        """获取数据库会话实例"""
        return cls.session

    @classmethod
    def close_session(cls):
        """关闭会话，释放数据库连接"""
        cls.session.remove()

2. 子类继承基类，快速操作指定表 #

比如原来的Action表操作类，可以简化为：

class ActionModel(DBBase):
    # 直接映射到已存在的actions表
    model = DBBase.Base.classes.actions

# 使用示例
if __name__ == "__main__":
    # 初始化模型
    action_model = ActionModel()
    # 获取会话
    session = action_model.get_session()

    # 执行查询操作
    all_actions = session.query(action_model.model).all()
    for action in all_actions:
        print(f"Action ID: {action.id}, Name: {action.name}")

    # 操作完成后关闭会话
    action_model.close_session()

关键优化点 #

• 复用性提升：所有数据库连接逻辑都在基类中实现，子类只需指定对应表即可，无需重复编写引擎、会话初始化代码
• 线程安全：使用scoped_session替代直接创建Session实例，保证多线程环境下的会话安全
• 资源管理：提供close_session方法，规范会话关闭流程，避免数据库连接泄漏
• 全局初始化：引擎和automap基类仅初始化一次，减少不必要的资源消耗

扩展说明 #

如果需要操作其他现有表，只需创建新的子类并指定对应表即可：

class UserModel(DBBase):
    model = DBBase.Base.classes.users

新的UserModel类同样可以直接使用基类的引擎、会话功能，无需重复编写连接逻辑。

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