Python Dependency Injector如何实现单请求共用单个SQLAlchemy DB连接
解决方案
核心思路是使用依赖注入框架的上下文作用域特性,将DB实例的生命周期绑定到单次请求上,实现单请求内所有依赖复用同一DB连接,不同请求使用独立连接。
1. 调整DatabaseContainer配置
替换原有的Factory+全局Singleton配置,将Database实例设为上下文作用域:
# DatabaseContainer.py from dependency_injector import containers, providers class DatabaseContainer(containers.DeclarativeContainer): # 上下文作用域:同一请求上下文内仅返回同一个Database实例 database = providers.ContextualScoped(Database) # 从上下文共享的Database实例中获取mysql session,单请求内复用 db_mysql = providers.Callable(database.provided.get_mysql_session) # 若需要psql则按相同逻辑添加 # db_psql = providers.Callable(database.provided.get_psql_session)
2. 修正AppContainer配置笔误
原AppContainer中ServiceContainer的repository参数传值错误,修正后:
# AppContainer.py class AppContainer(containers.DeclarativeContainer): database = providers.Container(DatabaseContainer) repository = providers.Container(RepositoryContainer, database=database) service = providers.Container( ServiceContainer, repository=repository, # 修正原crm_repository的笔误 database=database )
3. 调整请求生命周期逻辑
在单次请求执行前后添加上下文重置逻辑,确保每次请求创建新的DB连接,请求结束后销毁旧连接:
# 客户端代码 class OrderWatcherJob(BaseJob): def __init__(self): self._logger = logging.getLogger(__name__) self.func = self.execute_job @inject def execute_job( self, order_service: OrderService = Provide[AppContainer.service.order], db_mysql:Session = Provide[AppContainer.database.db_mysql] ): # 请求开始前重置上下文,本次请求将创建新的Database实例 AppContainer.reset_context() try: order_service.execute_something() db_mysql.commit() except Exception as ex: self._logger.error(ex) db_mysql.rollback() finally: db_mysql.close() # 请求结束后重置上下文,销毁本次的Database实例 AppContainer.reset_context()
验证效果
调整完成后,每次调用execute_job只会打印一次SESSION CREATED日志,说明单请求内仅创建了一个DB连接,OrderService、各Repository以及注入的db_mysql参数均复用该连接,满足事务ACID的要求。
内容的提问来源于stack exchange,提问作者Faris Dewantoro
相关产品推荐
相关产品推荐

