-
-
-DEFAULT_SESSION_FACTORY = sessionmaker(
- bind=create_engine(
- config.get_postgres_uri(),
- isolation_level="REPEATABLE READ",
- )
-)
-
-
-class SqlAlchemyUnitOfWork(AbstractUnitOfWork):
- def __init__(self, session_factory=DEFAULT_SESSION_FACTORY):
- self.session_factory = session_factory
-
- def __enter__(self):
- self.session = self.session_factory() # type: Session
- self.oclouds = ocloud_repository\
- .OcloudSqlAlchemyRepository(self.session)
- return super().__enter__()
-
- def __exit__(self, *args):
- super().__exit__(*args)
- self.session.close()
-
- def _commit(self):
- self.session.commit()
-
- def rollback(self):
- self.session.rollback()