getting there

Author: sanderegg

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

@@ -26,17 +26,23 @@ class LockLostError(BaseRedisError):
 ProjectLockError: TypeAlias = redis.exceptions.LockError  # NOTE: backwards compatible
 
 
-class SemaphoreAcquisitionError(BaseRedisError):
+class SemaphoreError(BaseRedisError):
+    msg_template: str = (
+        "Unexpected error with semaphore '{name}' by this instance `{instance_id}`"
+    )
+
+
+class SemaphoreAcquisitionError(SemaphoreError):
     msg_template: str = (
         "Could not acquire semaphore '{name}' by this instance `{instance_id}`"
     )
 
 
-class SemaphoreNotAcquiredError(BaseRedisError):
+class SemaphoreNotAcquiredError(SemaphoreError):
     msg_template: str = (
         "Semaphore '{name}' was not acquired by this instance `{instance_id}`"
     )
 
 
-class SemaphoreLostError(BaseRedisError):
+class SemaphoreLostError(SemaphoreError):
     msg_template: str = "Semaphore '{name}' was lost by this instance `{instance_id}`"

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

@@ -40,6 +40,7 @@
 )
 from ._errors import (
     SemaphoreAcquisitionError,
+    SemaphoreError,
     SemaphoreLostError,
     SemaphoreNotAcquiredError,
 )
@@ -452,10 +453,10 @@ async def _periodic_reacquisition(
         if not started.is_set():
             started.set()
 
+    lock_acquisition_time = arrow.utcnow()
     try:
         if not await semaphore.acquire():
             raise SemaphoreAcquisitionError(name=key, instance_id=semaphore.instance_id)
-        lock_acquisition_time = arrow.utcnow()
 
         async with (
             asyncio.TaskGroup() as tg
@@ -471,15 +472,13 @@ async def _periodic_reacquisition(
 
             await cancel_wait_task(auto_reacquisition_task)
     except BaseExceptionGroup as eg:
-        semaphore_lost_errors, other_errors = eg.split(
-            SemaphoreLostError | SemaphoreNotAcquiredError
-        )
+        semaphore_errors, other_errors = eg.split(SemaphoreError)
         if other_errors:
             assert len(other_errors.exceptions) == 1  # nosec
-            raise other_errors from eg
-        assert semaphore_lost_errors is not None  # nosec
-        assert len(semaphore_lost_errors.exceptions) == 1  # nosec
-        raise semaphore_lost_errors.exceptions[0] from eg
+            raise other_errors.exceptions[0] from eg
+        assert semaphore_errors is not None  # nosec
+        assert len(semaphore_errors.exceptions) == 1  # nosec
+        raise semaphore_errors.exceptions[0] from eg
     finally:
         try:
             await semaphore.release()

packages/service-library/tests/redis/test_semaphore.py

@@ -496,7 +496,7 @@ async def test_semaphore_context_manager_with_exception(
     captured_semaphore: DistributedSemaphore | None = None
 
     async def _raising_context():
-        async with DistributedSemaphore(
+        async with distributed_semaphore(
             redis_client=redis_client_sdk,
             key=semaphore_name,
             capacity=semaphore_capacity,
@@ -524,10 +524,10 @@ async def test_multiple_semaphores_different_keys(
     capacity = 1
 
     async with (
-        DistributedSemaphore(
+        distributed_semaphore(
             redis_client=redis_client_sdk, key=key1, capacity=capacity
         ),
-        DistributedSemaphore(
+        distributed_semaphore(
             redis_client=redis_client_sdk, key=key2, capacity=capacity
         ),
     ):