refactoring project lock

Author: matusdrobuliak66

packages/service-library/src/servicelib/project_lock.py

@@ -0,0 +1,78 @@
+import datetime
+import logging
+from asyncio.log import logger
+from collections.abc import AsyncIterator
+from contextlib import asynccontextmanager
+from typing import Final
+
+import redis
+from models_library.projects import ProjectID
+from models_library.projects_access import Owner
+from models_library.projects_state import ProjectLocked, ProjectStatus
+from redis.asyncio.lock import Lock
+from servicelib.background_task import periodic_task
+from servicelib.logging_utils import log_context
+
+_logger = logging.getLogger(__name__)
+
+PROJECT_REDIS_LOCK_KEY: str = "project_lock:{}"
+PROJECT_LOCK_TIMEOUT: Final[datetime.timedelta] = datetime.timedelta(seconds=10)
+ProjectLock = Lock
+
+ProjectLockError = redis.exceptions.LockError
+
+
+async def _auto_extend_project_lock(project_lock: Lock) -> None:
+    # NOTE: the background task already catches anything that might raise here
+    await project_lock.reacquire()
+
+
+@asynccontextmanager
+async def lock_project(
+    redis_lock: Lock,
+    project_uuid: str | ProjectID,
+    status: ProjectStatus,
+    owner: Owner | None = None,
+) -> AsyncIterator[None]:
+    """Context manager to lock and unlock a project by user_id
+
+    Raises:
+        ProjectLockError: if project is already locked
+    """
+
+    try:
+        if not await redis_lock.acquire(
+            blocking=False,
+            token=ProjectLocked(
+                value=True,
+                owner=owner,
+                status=status,
+            ).json(),
+        ):
+            msg = f"Lock for project {project_uuid!r} owner {owner!r} could not be acquired"
+            raise ProjectLockError(msg)
+
+        with log_context(
+            _logger,
+            logging.DEBUG,
+            msg=f"with lock for {owner=}:{project_uuid=}:{status=}",
+        ):
+            async with periodic_task(
+                _auto_extend_project_lock,
+                interval=0.6 * PROJECT_LOCK_TIMEOUT,
+                task_name=f"{PROJECT_REDIS_LOCK_KEY.format(project_uuid)}_lock_auto_extend",
+                project_lock=redis_lock,
+            ):
+                yield
+
+    finally:
+        # let's ensure we release that stuff
+        try:
+            if await redis_lock.owned():
+                await redis_lock.release()
+        except (redis.exceptions.LockError, redis.exceptions.LockNotOwnedError) as exc:
+            logger.warning(
+                "releasing %s unexpectedly raised an exception: %s",
+                f"{redis_lock=!r}",
+                f"{exc}",
+            )

services/director-v2/tests/unit/test_utils_distributed_identifier.py

@@ -163,7 +163,6 @@ async def _destroy(
         self.api.delete(identifier)
 
 
-# MD: here redis
 @pytest.fixture
 async def redis_client_sdk(
     redis_service: RedisSettings,

services/efs-guardian/src/simcore_service_efs_guardian/services/background_tasks.py

@@ -1,15 +1,14 @@
 import logging
-from contextlib import asynccontextmanager
-from datetime import datetime, timedelta, timezone
-from typing import AsyncIterator, Final
+from datetime import datetime, timezone
 
-import redis
 from fastapi import FastAPI
 from models_library.projects import ProjectID
-from models_library.projects_state import ProjectLocked, ProjectStatus
-from redis.asyncio.lock import Lock
-from servicelib.background_task import periodic_task
-from servicelib.logging_utils import log_context
+from models_library.projects_state import ProjectStatus
+from servicelib.project_lock import (
+    PROJECT_LOCK_TIMEOUT,
+    PROJECT_REDIS_LOCK_KEY,
+    lock_project,
+)
 from simcore_postgres_database.utils_projects import ProjectsRepo
 
 from ..core.settings import ApplicationSettings
@@ -19,68 +18,6 @@
 _logger = logging.getLogger(__name__)
 
 
-PROJECT_REDIS_LOCK_KEY: str = "project_lock:{}"
-PROJECT_LOCK_TIMEOUT: Final[timedelta] = timedelta(seconds=10)
-
-
-async def _auto_extend_project_lock(project_lock: Lock) -> None:
-    # NOTE: the background task already catches anything that might raise here
-    await project_lock.reacquire()
-
-
-@asynccontextmanager
-async def lock_project(
-    app: FastAPI,
-    project_uuid: ProjectID,
-    status: ProjectStatus = ProjectStatus.MAINTAINING,
-) -> AsyncIterator[None]:
-    """Context manager to lock and unlock a project by user_id
-
-    Raises:
-        ProjectLockError: if project is already locked
-    """
-
-    redis_lock = get_redis_lock_client(app).redis.lock(
-        PROJECT_REDIS_LOCK_KEY.format(project_uuid),
-        timeout=PROJECT_LOCK_TIMEOUT.total_seconds(),
-    )
-    try:
-        if not await redis_lock.acquire(
-            blocking=False,
-            token=ProjectLocked(
-                value=True,
-                owner=None,
-                status=status,
-            ).json(),
-        ):
-            msg = f"Lock for project {project_uuid!r} could not be acquired"
-            raise ValueError(msg)
-
-        with log_context(
-            _logger,
-            logging.DEBUG,
-            msg=f"with lock for {project_uuid=}:{status=}",
-        ):
-            async with periodic_task(
-                _auto_extend_project_lock,
-                interval=0.6 * PROJECT_LOCK_TIMEOUT,
-                task_name=f"{PROJECT_REDIS_LOCK_KEY.format(project_uuid)}_lock_auto_extend",
-                project_lock=redis_lock,
-            ):
-                yield
-
-    finally:
-        try:
-            if await redis_lock.owned():
-                await redis_lock.release()
-        except (redis.exceptions.LockError, redis.exceptions.LockNotOwnedError) as exc:
-            _logger.warning(
-                "releasing %s unexpectedly raised an exception: %s",
-                f"{redis_lock=!r}",
-                f"{exc}",
-            )
-
-
 async def removal_policy_task(app: FastAPI) -> None:
     _logger.info("Removal policy task started")
 
@@ -111,5 +48,11 @@ async def removal_policy_task(app: FastAPI) -> None:
                 _project_last_change_date,
                 app_settings.EFS_REMOVAL_POLICY_TASK_AGE_LIMIT_TIMEDELTA,
             )
-            async with lock_project(app, project_uuid=project_id):
+            redis_lock = get_redis_lock_client(app).redis.lock(
+                PROJECT_REDIS_LOCK_KEY.format(project_id),
+                timeout=PROJECT_LOCK_TIMEOUT.total_seconds(),
+            )
+            async with lock_project(
+                redis_lock, project_uuid=project_id, status=ProjectStatus.MAINTAINING
+            ):
                 await efs_manager.remove_project_efs_data(project_id)

services/efs-guardian/tests/unit/test_api_health.py

@@ -24,7 +24,6 @@ def app_environment(
         monkeypatch,
         {
             **app_environment,
-            # **rabbit_env_vars_dict,
         },
     )