use-diff-sessions (#380)

Author: JKL98ISR

.github/workflows/license-check.yml

@@ -50,7 +50,7 @@ jobs:
         with:
           requirements: "backend/requirements-all.txt"
           fail: "Copyleft,Other,Error"
-          exclude: '(envier.*0\.5\.0|psycopg2.*2\.9\.3|fqdn.*1\.5\.1|pyzmq.*25\.1\.2|debugpy.*1\.6\.7|certifi.*2024\.8\.30|tqdm.*4\.66\..*|webencodings.*0\.5\.1|torch.*1\.10\.2.*|torch.*1\.11\.0.*|pytorch-ignite.*0\.4\.10.*|torchaudio.*0\.11\.0.*|torchvision.*0\.12\.0.*|terminado.*0\.15\.0|qudida.*0\.0\.4|expiringdict.*1\.2\.2|botocore.*1\.29\.80|orderedmultidict.*1\.0\.1|deepchecks.*)'
+          exclude: '(envier.*0\.5\.0|psycopg2.*2\.9\.3|fqdn.*1\.5\.1|pyzmq.*25\.1\.2|debugpy.*1\.6\.7|certifi.*2024\.8\.30|tqdm.*4\.67\..*|webencodings.*0\.5\.1|torch.*1\.10\.2.*|torch.*1\.11\.0.*|pytorch-ignite.*0\.4\.10.*|torchaudio.*0\.11\.0.*|torchvision.*0\.12\.0.*|terminado.*0\.15\.0|qudida.*0\.0\.4|expiringdict.*1\.2\.2|botocore.*1\.29\.80|orderedmultidict.*1\.0\.1|deepchecks.*)'
           # psycopg2 is LGPL 2
           # pyzmq is Revised BSD https://github.com/zeromq/pyzmq/blob/main/examples/LICENSE
           # debugpy is MIT https://github.com/microsoft/debugpy/blob/main/LICENSE

backend/deepchecks_monitoring/resources.py

@@ -26,7 +26,7 @@
 from sqlalchemy import select
 from sqlalchemy.ext.asyncio import AsyncEngine, create_async_engine
 from sqlalchemy.future.engine import Engine, create_engine
-from sqlalchemy.orm import Session, sessionmaker
+from sqlalchemy.orm import Session
 
 from deepchecks_monitoring import config
 from deepchecks_monitoring.features_control import FeaturesControl
@@ -79,9 +79,7 @@ class ResourcesProvider(BaseResourcesProvider):
     def __init__(self, settings: config.BaseSettings):
         self._settings = settings
         self._database_engine: t.Optional[Engine] = None
-        self._session_factory: t.Optional[sessionmaker] = None
         self._async_database_engine: t.Optional[AsyncEngine] = None
-        self._async_session_factory: t.Optional[sessionmaker] = None
         self._kafka_admin: t.Optional[KafkaAdminClient] = None
         self._redis_client: t.Optional[Redis] = None
         self._cache_funcs: t.Optional[CacheFunctions] = None
@@ -147,15 +145,11 @@ def settings(self) -> config.Settings:
 
     def dispose_resources(self):
         """Dispose resources."""
-        if self._session_factory is not None:
-            self._session_factory.close_all()
         if self._database_engine is not None:
             self._database_engine.dispose()
 
     async def async_dispose_resources(self):
         """Dispose async resources."""
-        # if self._async_session_factory is not None:
-        #     await AsyncSession.close_all()
         if self._async_database_engine is not None:
             await self._async_database_engine.dispose()
 
@@ -180,23 +174,15 @@ def database_engine(self) -> Engine:
 
         return self._database_engine
 
-    @property
-    def session_factory(self) -> sessionmaker:
-        """Return alchemy session factory."""
-        if self._session_factory is None:
-            self._session_factory = sessionmaker(
-                self.database_engine,
-                # class_=ExtendedAsyncSession,  # TODO:
-                autoflush=False,
-                expire_on_commit=False,
-                autocommit=False,
-            )
-        return self._session_factory
-
     @contextmanager
     def create_database_session(self) -> t.Iterator[Session]:
         """Create sqlalchemy database session."""
-        with self.session_factory() as session:  # pylint: disable=not-callable
+        with Session(
+            self.database_engine,
+            autoflush=False,
+            expire_on_commit=False,
+            autocommit=False,
+        ) as session:  # pylint: disable=not-callable
             try:
                 yield session
                 session.commit()
@@ -225,19 +211,6 @@ def async_database_engine(self) -> AsyncEngine:
         )
         return self._async_database_engine
 
-    @property
-    def async_session_factory(self) -> sessionmaker:
-        """Return async alchemy session maker."""
-        if self._async_session_factory is None:
-            self._async_session_factory = sessionmaker(
-                self.async_database_engine,
-                class_=ExtendedAsyncSession,
-                autoflush=False,
-                expire_on_commit=False,
-                autocommit=False,
-            )
-        return self._async_session_factory
-
     @t.overload
     def create_async_database_session(
         self,
@@ -260,8 +233,12 @@ async def create_async_database_session(
         organization_id: t.Optional[int] = None
     ) -> t.AsyncIterator[t.Optional[ExtendedAsyncSession]]:
         """Create async sqlalchemy database session."""
-        async with self.async_session_factory() as session:  # pylint: disable=not-callable
-            session: ExtendedAsyncSession
+        async with ExtendedAsyncSession(
+            self.async_database_engine,
+            autoflush=False,
+            expire_on_commit=False,
+            autocommit=False,
+        ) as session:  # pylint: disable=not-callable
             try:
                 if organization_id:
                     organization_schema = await session.scalar(

backend/tests/logic/test_cache_functions.py

@@ -37,7 +37,7 @@ async def test_clear_monitor_cache(resources_provider):
 
 
 @pytest.mark.asyncio
-async def test_delete_monitor_cache_by_timestamp(resources_provider):
+async def test_delete_monitor_cache_by_timestamp(resources_provider, async_session):
     cache_funcs: CacheFunctions = resources_provider.cache_functions
 
     # Arrange - Organization with 2 monitors and 2 model versions, and another organization with same monitor id.
@@ -60,7 +60,7 @@ async def test_delete_monitor_cache_by_timestamp(resources_provider):
     cache_funcs.add_invalidation_timestamps(1, 1, timestamps_to_invalidate)
 
     # Act - run task
-    async with resources_provider.async_session_factory() as session:
+    async with async_session as session:
         task_id = await insert_model_version_cache_invalidation_task(1, 1, session=session)
         task = await session.scalar(select(Task).where(Task.id == task_id))
         await ModelVersionCacheInvalidation().run(task, session, resources_provider, lock=None)