Python Depedency Injector passing single SQLAlchemy db connection every request

Viewed 291

I try to implement DI in my console app using depedency_injector

I have some rules that every request should use the same DB Connection to all dependencies, so when there is a request it will create a new DB Connection, 1 DbConnection each request so I can make this transactional (ACID).

Here is my code, is still create a new DB connection on all dependencies. I don't want to use Singleton since it only used 1 connection.

Database.py

engine_mysql = create_engine(DB_MYSQL_DSN, pool_size=20, max_overflow=0)
session_mysql = sessionmaker(autocommit=False, autoflush=False, bind=engine_mysql)

class Database:
    _session_mysql: Session
    _session_psql: Session

    def __init__(self) -> None:
        print(100 * "=")
        print("SESSION CREATED")
        print(100 * "=")
        self._session_mysql = session_mysql()

    def get_mysql_session(self):
        print(100 * "=")
        print("SESSION CALL MYSQL")
        print(100 * "=")
        return self._session_mysql

DatabaseContainer.py

class DatabaseContainer(containers.DeclarativeContainer):

    database = providers.Factory(Database)
    db_mysql = providers.Singleton(database.provided.get_mysql_session)

RepositoryContainer.py

class RepositoryContainer(containers.DeclarativeContainer):
    database = providers.DependenciesContainer()
    transaction = providers.Factory(TransactionRepository, session=database.db_mysql)
    transaction_package = providers.Factory(
      TransactionPackageRepository,
      session=database.db_psql,
    )

ServiceContainer.py

class ServiceContainer(containers.DeclarativeContainer):
    repository = providers.DependenciesContainer()
    database = providers.DependenciesContainer()

    order = providers.Factory(
        orderService,
        # database
        db_conn=database.db_mysql,
        # repository
        transaction_repo=repository.transaction,
        transaction_package_repo=repository.transaction_package,
    )

AppContainer.py

class AppContainer(containers.DeclarativeContainer):

    database = providers.Container(DatabaseContainer)

    repository = providers.Container(RepositoryContainer, database=database)
    
    service = providers.Container(
        ServiceContainer,
        repository=crm_repository,
        database=database
    )

Client Code

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]
    ):

        try:
            order_service.execute_something()
            db_mysql.commit()
            #or order_service.db_conn.commit() 
        except Exception as ex:
            self._logger.error(ex)
            db_mysql.rollback()
            #or order_service.db_conn.rollback() 
        finally:
            db_mysql.close()
           #or order_service.db_conn.close() 

So.. how do I make this become used only 1 DB Connection every request ? I mean when I inject my OrderService it will use the same DB Connection on TransactionRepo, TransactionPackageRepo, and also db_conn.

0 Answers
Related