refactored redis client structure

Author: sanderegg

packages/pytest-simcore/src/pytest_simcore/redis_service.py

@@ -10,6 +10,7 @@
 import tenacity
 from pytest_mock import MockerFixture
 from redis.asyncio import Redis, from_url
+from servicelib.redis import _constants as redis_constants
 from settings_library.basic_types import PortInt
 from settings_library.redis import RedisDatabase, RedisSettings
 from tenacity.before_sleep import before_sleep_log
@@ -118,6 +119,4 @@ async def wait_till_redis_responsive(redis_url: URL | str) -> None:
 @pytest.fixture
 def mock_redis_socket_timeout(mocker: MockerFixture) -> None:
     # lowered to allow CI to properly shutdown RedisClientSDK instances
-    from servicelib import redis
-
-    mocker.patch.object(redis, "_DEFAULT_SOCKET_TIMEOUT", timedelta(seconds=1))
+    mocker.patch.object(redis_constants, "DEFAULT_SOCKET_TIMEOUT", timedelta(seconds=1))

packages/service-library/src/servicelib/redis/__init__.py

@@ -1,14 +1,23 @@
-from ._client import RabbitMQClient
+from ._client import RedisClientSDK
+from ._clients_manager import RedisClientsManager
+from ._decorators import exclusive
+from ._distributed_locks_utils import start_exclusive_periodic_task
 from ._errors import (
     CouldNotAcquireLockError,
     CouldNotConnectToRedisError,
     LockLostError,
 )
+from ._models import RedisManagerDBConfig
 
 __all__: tuple[str, ...] = (
     "CouldNotAcquireLockError",
     "CouldNotConnectToRedisError",
+    "exclusive",
     "LockLostError",
+    "RedisClientSDK",
+    "RedisClientsManager",
+    "RedisManagerDBConfig",
+    "start_exclusive_periodic_task",
 )
 
 # nopycln: file

packages/service-library/src/servicelib/redis/_client.py

@@ -47,7 +47,7 @@ class RedisClientSDK:
     def redis(self) -> aioredis.Redis:
         return self._client
 
-    def __post_init__(self):
+    def __post_init__(self) -> None:
         self._client = aioredis.from_url(
             self.redis_dsn,
             # Run 3 retries with exponential backoff strategy source: https://redis.readthedocs.io/en/stable/backoff.html

packages/service-library/src/servicelib/redis/_distributed_locks_utils.py

@@ -0,0 +1,79 @@
+import asyncio
+import datetime
+import logging
+from collections.abc import Awaitable, Callable
+
+import arrow
+from servicelib.background_task import start_periodic_task
+
+from ._client import RedisClientSDK
+from ._decorators import exclusive
+from ._errors import CouldNotAcquireLockError
+
+_logger = logging.getLogger(__name__)
+
+
+async def _exclusive_task_starter(
+    client: RedisClientSDK,
+    usr_tsk_task: Callable[..., Awaitable[None]],
+    *,
+    usr_tsk_interval: datetime.timedelta,
+    usr_tsk_task_name: str,
+    **kwargs,
+) -> None:
+    lock_key = f"lock:exclusive_task_starter:{usr_tsk_task_name}"
+    lock_value = f"locked since {arrow.utcnow().format()}"
+
+    try:
+        await exclusive(client, lock_key=lock_key, lock_value=lock_value)(
+            start_periodic_task
+        )(
+            usr_tsk_task,
+            interval=usr_tsk_interval,
+            task_name=usr_tsk_task_name,
+            **kwargs,
+        )
+    except CouldNotAcquireLockError:
+        _logger.debug(
+            "Could not acquire lock '%s' with value '%s'", lock_key, lock_value
+        )
+    except Exception as e:
+        _logger.exception(e)  # noqa: TRY401
+        raise
+
+
+def start_exclusive_periodic_task(
+    client: RedisClientSDK,
+    task: Callable[..., Awaitable[None]],
+    *,
+    task_period: datetime.timedelta,
+    retry_after: datetime.timedelta = datetime.timedelta(seconds=1),
+    task_name: str,
+    **kwargs,
+) -> asyncio.Task:
+    """
+    Ensures that only 1 process periodically ever runs ``task`` at all times.
+    If one process dies, another process will run the ``task``.
+
+    Creates a background task that periodically tries to start the user ``task``.
+    Before the ``task`` is scheduled for periodic background execution, it acquires a lock.
+    Subsequent calls to ``start_exclusive_periodic_task`` will not allow the same ``task``
+    to start since the lock will prevent the scheduling.
+
+    Q&A:
+    - Why is `_exclusive_task_starter` run as a task?
+        This is usually used at setup time and cannot block the setup process forever
+    - Why is `_exclusive_task_starter` task a periodic task?
+        If Redis connectivity is lost, the periodic `_exclusive_task_starter` ensures the lock is
+        reacquired
+    """
+    return start_periodic_task(
+        _exclusive_task_starter,
+        interval=retry_after,
+        task_name=f"exclusive_task_starter_{task_name}",
+        client=client,
+        usr_tsk_task=task,
+        usr_tsk_interval=task_period,
+        usr_tsk_task_name=task_name,
+        **kwargs,
+    )

packages/service-library/src/servicelib/redis/_utils.py

@@ -1,19 +1,14 @@
 import asyncio
-import datetime
 import logging
-from collections.abc import Awaitable, Callable
+from collections.abc import Awaitable
 from typing import Any
 
-import arrow
 import redis.exceptions
 from redis.asyncio.lock import Lock
-from servicelib.background_task import start_periodic_task
 
 from ..logging_utils import log_context
-from ._client import RedisClientSDK
 from ._constants import SHUTDOWN_TIMEOUT_S
-from ._decorators import exclusive
-from ._errors import CouldNotAcquireLockError, LockLostError
+from ._errors import LockLostError
 
 _logger = logging.getLogger(__name__)
 
@@ -29,78 +24,12 @@ async def cancel_or_warn(task: asyncio.Task) -> None:
 
 async def auto_extend_lock(lock: Lock) -> None:
     try:
-        with log_context(_logger, logging.DEBUG, f"Autoextend lock {lock.name}"):
+        with log_context(_logger, logging.DEBUG, f"Autoextend lock {lock.name!r}"):
             await lock.reacquire()
     except redis.exceptions.LockNotOwnedError as exc:
         raise LockLostError(lock=lock) from exc
 
 
-async def _exclusive_task_starter(
-    client: RedisClientSDK,
-    usr_tsk_task: Callable[..., Awaitable[None]],
-    *,
-    usr_tsk_interval: datetime.timedelta,
-    usr_tsk_task_name: str,
-    **kwargs,
-) -> None:
-    lock_key = f"lock:exclusive_task_starter:{usr_tsk_task_name}"
-    lock_value = f"locked since {arrow.utcnow().format()}"
-
-    try:
-        await exclusive(client, lock_key=lock_key, lock_value=lock_value)(
-            start_periodic_task
-        )(
-            usr_tsk_task,
-            interval=usr_tsk_interval,
-            task_name=usr_tsk_task_name,
-            **kwargs,
-        )
-    except CouldNotAcquireLockError:
-        _logger.debug(
-            "Could not acquire lock '%s' with value '%s'", lock_key, lock_value
-        )
-    except Exception as e:
-        _logger.exception(e)  # noqa: TRY401
-        raise
-
-
-def start_exclusive_periodic_task(
-    client: RedisClientSDK,
-    task: Callable[..., Awaitable[None]],
-    *,
-    task_period: datetime.timedelta,
-    retry_after: datetime.timedelta = datetime.timedelta(seconds=1),
-    task_name: str,
-    **kwargs,
-) -> asyncio.Task:
-    """
-    Ensures that only 1 process periodically ever runs ``task`` at all times.
-    If one process dies, another process will run the ``task``.
-
-    Creates a background task that periodically tries to start the user ``task``.
-    Before the ``task`` is scheduled for periodic background execution, it acquires a lock.
-    Subsequent calls to ``start_exclusive_periodic_task`` will not allow the same ``task``
-    to start since the lock will prevent the scheduling.
-
-    Q&A:
-    - Why is `_exclusive_task_starter` run as a task?
-        This is usually used at setup time and cannot block the setup process forever
-    - Why is `_exclusive_task_starter` task a periodic task?
-        If Redis connectivity is lost, the periodic `_exclusive_task_starter` ensures the lock is
-        reacquired
-    """
-    return start_periodic_task(
-        _exclusive_task_starter,
-        interval=retry_after,
-        task_name=f"exclusive_task_starter_{task_name}",
-        client=client,
-        usr_tsk_task=task,
-        usr_tsk_interval=task_period,
-        usr_tsk_task_name=task_name,
-        **kwargs,
-    )
-
-
 async def handle_redis_returns_union_types(result: Any | Awaitable[Any]) -> Any:
     """Used to handle mypy issues with redis 5.x return types"""
     if isinstance(result, Awaitable):

packages/service-library/tests/conftest.py

@@ -77,7 +77,8 @@ async def get_redis_client_sdk(
 ]:
     @asynccontextmanager
     async def _(
-        database: RedisDatabase, decode_response: bool = True  # noqa: FBT002
+        database: RedisDatabase,
+        decode_response: bool = True,  # noqa: FBT002
     ) -> AsyncIterator[RedisClientSDK]:
         redis_resources_dns = redis_service.build_redis_dsn(database)
         client = RedisClientSDK(
@@ -97,7 +98,7 @@ async def _cleanup_redis_data(clients_manager: RedisClientsManager) -> None:
             await clients_manager.client(db).redis.flushall()
 
     async with RedisClientsManager(
-        {RedisManagerDBConfig(db) for db in RedisDatabase},
+        {RedisManagerDBConfig(database=db) for db in RedisDatabase},
         redis_service,
         client_name="pytest",
     ) as clients_manager:

packages/service-library/tests/deferred_tasks/test_deferred_tasks.py

@@ -20,9 +20,9 @@
 from common_library.serialization import model_dump_with_secrets
 from pydantic import NonNegativeFloat, NonNegativeInt
 from pytest_mock import MockerFixture
-from servicelib import redis as servicelib_redis
 from servicelib.rabbitmq import RabbitMQClient
 from servicelib.redis import RedisClientSDK
+from servicelib.redis import _constants as redis_client_constants
 from servicelib.sequences_utils import partition_gen
 from settings_library.rabbit import RabbitSettings
 from settings_library.redis import RedisSettings
@@ -353,7 +353,6 @@ def __init__(
     async def _pause_container(
         self, container_name: str, client: ClientWithPingProtocol
     ) -> AsyncIterator[None]:
-
         async with self.paused_container(container_name):
             async for attempt in AsyncRetrying(
                 wait=wait_fixed(0.1),
@@ -391,7 +390,9 @@ async def pause_redis(self) -> AsyncIterator[None]:
 @pytest.fixture
 def mock_default_socket_timeout(mocker: MockerFixture) -> None:
     mocker.patch.object(
-        servicelib_redis, "_DEFAULT_SOCKET_TIMEOUT", datetime.timedelta(seconds=0.25)
+        redis_client_constants,
+        "DEFAULT_SOCKET_TIMEOUT",
+        datetime.timedelta(seconds=0.25),
     )
 
 
@@ -420,7 +421,6 @@ async def test_workflow_with_third_party_services_outages(
         redis_service,
         max_workers,
     ) as manager:
-
         # start all in parallel
         await asyncio.gather(
             *[manager.start_task(0.1, i) for i in range(deferred_tasks_to_start)]

packages/service-library/tests/test_redis.py

@@ -14,14 +14,14 @@
 from faker import Faker
 from pytest_mock import MockerFixture
 from redis.exceptions import LockError, LockNotOwnedError
-from servicelib import redis as servicelib_redis
 from servicelib.redis import (
     CouldNotAcquireLockError,
     LockLostError,
     RedisClientSDK,
     RedisClientsManager,
     RedisManagerDBConfig,
 )
+from servicelib.redis import _constants as redis_constants
 from servicelib.utils import limited_gather
 from settings_library.redis import RedisDatabase, RedisSettings
 from tenacity import (
@@ -63,7 +63,7 @@ def lock_timeout() -> datetime.timedelta:
 @pytest.fixture
 def mock_default_lock_ttl(mocker: MockerFixture) -> None:
     mocker.patch.object(
-        servicelib_redis, "_DEFAULT_LOCK_TTL", datetime.timedelta(seconds=0.25)
+        redis_constants, "DEFAULT_LOCK_TTL", datetime.timedelta(seconds=0.25)
     )
 
 
@@ -166,6 +166,7 @@ async def test_lock_context(
     assert await ttl_lock.owned() is False
 
 
+@pytest.mark.xfail(reason="This test shows an issue, that will be fixed in the next PR")
 async def test_lock_context_raises_if_lock_is_lost(
     redis_client_sdk: RedisClientSDK, faker: Faker
 ):
@@ -180,7 +181,6 @@ async def test_lock_context_raises_if_lock_is_lost(
             await asyncio.sleep(20)
 
 
-@pytest.mark.xfail(reason="This test shows an issue, that will be fixed in the next PR")
 async def test_lock_context_with_already_locked_lock_raises(
     redis_client_sdk: RedisClientSDK, faker: Faker
 ):
@@ -213,7 +213,7 @@ async def test_lock_context_with_data(redis_client_sdk: RedisClientSDK, faker: F
     lock_name = faker.pystr()
     assert await _is_locked(redis_client_sdk, lock_name) is False
     assert await redis_client_sdk.lock_value(lock_name) is None
-    async with redis_client_sdk.lock_context(lock_name, lock_value=lock_data) as lock:
+    async with redis_client_sdk.lock_context(lock_name, lock_value=lock_data):
         assert await _is_locked(redis_client_sdk, lock_name) is True
         assert await redis_client_sdk.lock_value(lock_name) == lock_data
     assert await _is_locked(redis_client_sdk, lock_name) is False
@@ -254,18 +254,14 @@ async def race_condition_increase(self, by: int) -> None:
             current_value = self.value
             current_value += by
             # most likely situation which creates issues
-            await asyncio.sleep(
-                servicelib_redis._DEFAULT_LOCK_TTL.total_seconds() / 2  # noqa: SLF001
-            )
+            await asyncio.sleep(redis_constants.DEFAULT_LOCK_TTL.total_seconds() / 2)
             self.value = current_value
 
     counter = RaceConditionCounter()
     lock_name: str = faker.pystr()
     # ensures it does nto time out before acquiring the lock
     time_for_all_inc_counter_calls_to_finish_s: float = (
-        servicelib_redis._DEFAULT_LOCK_TTL.total_seconds()  # noqa: SLF001
-        * INCREASE_OPERATIONS
-        * 10
+        redis_constants.DEFAULT_LOCK_TTL.total_seconds() * INCREASE_OPERATIONS * 10
     )
 
     async def _inc_counter() -> None:
@@ -289,7 +285,7 @@ async def test_redis_client_sdks_manager(
     mock_redis_socket_timeout: None, redis_service: RedisSettings
 ):
     all_redis_configs: set[RedisManagerDBConfig] = {
-        RedisManagerDBConfig(db) for db in RedisDatabase
+        RedisManagerDBConfig(database=db) for db in RedisDatabase
     }
     manager = RedisClientsManager(
         databases_configs=all_redis_configs,
@@ -335,7 +331,7 @@ async def test_redis_client_sdk_setup_shutdown(
 @pytest.fixture
 def mock_default_socket_timeout(mocker: MockerFixture) -> None:
     mocker.patch.object(
-        servicelib_redis, "_DEFAULT_SOCKET_TIMEOUT", datetime.timedelta(seconds=0.25)
+        redis_constants, "DEFAULT_SOCKET_TIMEOUT", datetime.timedelta(seconds=0.25)
     )