Replacing Redis with Cache

Author: leandrodamascena

Makefile

@@ -7,18 +7,18 @@ target:
 dev:
 	pip install --upgrade pip pre-commit poetry
 	@$(MAKE) dev-version-plugin
-	poetry install --extras "all redis datamasking"
+	poetry install --extras "all redis datamasking valkey"
 	pre-commit install
 
 dev-quality-code:
 	pip install --upgrade pip pre-commit poetry
 	@$(MAKE) dev-version-plugin
-	poetry install --extras "all redis datamasking"
+	poetry install --extras "all redis datamasking valkey"
 	pre-commit install
 
 dev-gitpod:
 	pip install --upgrade pip poetry
-	poetry install --extras "all redis datamasking"
+	poetry install --extras "all redis datamasking valkey"
 	pre-commit install
 
 # Running licensecheck with zero to break the pipeline if there is an invalid license

aws_lambda_powertools/utilities/idempotency/persistence/cache.py

@@ -0,0 +1,11 @@
+from aws_lambda_powertools.utilities.idempotency.persistence.redis import (
+    CacheClientProtocol,
+    CacheConnection,
+    CachePersistenceLayer,
+)
+
+__all__ = [
+    "CacheClientProtocol",
+    "CachePersistenceLayer",
+    "CacheConnection",
+]

aws_lambda_powertools/utilities/idempotency/persistence/redis.py

@@ -8,7 +8,7 @@
 from typing import Any, Literal, Protocol
 
 import redis
-from typing_extensions import deprecated
+from typing_extensions import TypeAlias, deprecated
 
 from aws_lambda_powertools.utilities.idempotency import BasePersistenceLayer
 from aws_lambda_powertools.utilities.idempotency.exceptions import (
@@ -88,7 +88,7 @@ def __init__(
         host: str = "",
         port: int = 6379,
         username: str = "",
-        password: str = "",  # nosec - password for Redis connection
+        password: str = "",  # nosec - password for Cache connection
         db_index: int = 0,
         mode: Literal["standalone", "cluster"] = "standalone",
         ssl: bool = True,
@@ -131,7 +131,7 @@ def __init__(
 
         from aws_lambda_powertools.utilities.typing import LambdaContext
 
-        persistence_layer = RedisCachePersistenceLayer(host="localhost", port=6379)
+        persistence_layer = CachePersistenceLayer(host="localhost", port=6379)
 
 
         @dataclass
@@ -184,15 +184,15 @@ def _init_client(self) -> RedisClientProtocol:
 
         try:
             if self.url:
-                logger.debug(f"Using URL format to connect to Redis: {self.host}")
+                logger.debug(f"Using URL format to connect to Cache: {self.host}")
                 return client.from_url(url=self.url)
             else:
-                # Redis in cluster mode doesn't support db parameter
+                # Cache in cluster mode doesn't support db parameter
                 extra_param_connection: dict[str, Any] = {}
                 if self.mode != "cluster":
                     extra_param_connection = {"db": self.db_index}
 
-                logger.debug(f"Using arguments to connect to Redis: {self.host}")
+                logger.debug(f"Using arguments to connect to Cache: {self.host}")
                 return client(
                     host=self.host,
                     port=self.port,
@@ -203,8 +203,8 @@ def _init_client(self) -> RedisClientProtocol:
                     **extra_param_connection,
                 )
         except redis.exceptions.ConnectionError as exc:
-            logger.debug(f"Cannot connect in Redis: {self.host}")
-            raise IdempotencyPersistenceConnectionError("Could not to connect to Redis", exc) from exc
+            logger.debug(f"Cannot connect to Cache endpoint: {self.host}")
+            raise IdempotencyPersistenceConnectionError("Could not to connect to Cache endpoint", exc) from exc
 
 
 @deprecated("RedisCachePersistenceLayer will be removed in v4.0.0. Please use CachePersistenceLayer instead.")
@@ -215,7 +215,7 @@ def __init__(
         host: str = "",
         port: int = 6379,
         username: str = "",
-        password: str = "",  # nosec - password for Redis connection
+        password: str = "",  # nosec - password for Cache connection
         db_index: int = 0,
         mode: Literal["standalone", "cluster"] = "standalone",
         ssl: bool = True,
@@ -227,39 +227,39 @@ def __init__(
         validation_key_attr: str = "validation",
     ):
         """
-        Initialize the Redis Persistence Layer
+        Initialize the Cache Persistence Layer
 
         Parameters
         ----------
         host: str, optional
-            Redis host
+            Cache host
         port: int, optional: default 6379
-            Redis port
+            Cache port
         username: str, optional
-            Redis username
+            Cache username
         password: str, optional
-            Redis password
+            Cache password
         url: str, optional
-            Redis connection string, using url will override the host/port in the previous parameters
+            Cache connection string, using url will override the host/port in the previous parameters
         db_index: int, optional: default 0
-            Redis db index
+            Cache db index
         mode: str, Literal["standalone","cluster"]
-            set Redis client mode, choose from standalone/cluster
+            set Cache client mode, choose from standalone/cluster
         ssl: bool, optional: default True
-            set whether to use ssl for Redis connection
-        client: RedisClientProtocol, optional
-            Bring your own Redis client that follows RedisClientProtocol.
+            set whether to use ssl for Cache connection
+        client: CacheClientProtocol, optional
+            Bring your own Cache client that follows CacheClientProtocol.
             If provided, all other connection configuration options will be ignored
         expiry_attr: str, optional
-            Redis json attribute name for expiry timestamp, by default "expiration"
+            Cache json attribute name for expiry timestamp, by default "expiration"
         in_progress_expiry_attr: str, optional
-            Redis json attribute name for in-progress expiry timestamp, by default "in_progress_expiration"
+            Cache json attribute name for in-progress expiry timestamp, by default "in_progress_expiration"
         status_attr: str, optional
-            Redis json attribute name for status, by default "status"
+            Cache json attribute name for status, by default "status"
         data_attr: str, optional
-            Redis json attribute name for response data, by default "data"
+            Cache json attribute name for response data, by default "data"
         validation_key_attr: str, optional
-            Redis json attribute name for hashed representation of the parts of the event used for validation
+            Cache json attribute name for hashed representation of the parts of the event used for validation
 
         Examples
         --------
@@ -270,16 +270,16 @@ def __init__(
             idempotent,
         )
 
-        from aws_lambda_powertools.utilities.idempotency.persistence.redis import (
-            RedisCachePersistenceLayer,
+        from aws_lambda_powertools.utilities.idempotency.persistence.cache import (
+            CachePersistenceLayer,
         )
 
         client = redis.Redis(
             host="localhost",
             port="6379",
             decode_responses=True,
         )
-        persistence_layer = RedisCachePersistenceLayer(client=client)
+        persistence_layer = CachePersistenceLayer(client=client)
 
         @idempotent(persistence_store=persistence_layer)
         def lambda_handler(event: dict, context: LambdaContext):
@@ -292,7 +292,7 @@ def lambda_handler(event: dict, context: LambdaContext):
         ```
         """
 
-        # Initialize Redis client with Redis config if no client is passed in
+        # Initialize Cache client with cache config if no client is passed in
         if client is None:
             self.client = RedisConnection(
                 host=host,
@@ -334,11 +334,11 @@ def _item_to_data_record(self, idempotency_key: str, item: dict[str, Any]) -> Da
             in_progress_expiry_timestamp=in_progress_expiry_timestamp,
             response_data=str(item.get(self.data_attr)),
             payload_hash=str(item.get(self.validation_key_attr)),
-            expiry_timestamp=item.get("expiration", None),
+            expiry_timestamp=item.get("expiration"),
         )
 
     def _get_record(self, idempotency_key) -> DataRecord:
-        # See: https://redis.io/commands/get/
+        # See: https://valkey.io/valkey-glide/python/core/#glide.async_commands.CoreCommands.set
         response = self.client.get(idempotency_key)
 
         # key not found
@@ -388,25 +388,25 @@ def _put_in_progress_record(self, data_record: DataRecord) -> None:
             # The idempotency key does not exist:
             #    - first time that this invocation key is used
             #    - previous invocation with the same key was deleted due to TTL
-            #    - SET see https://redis.io/commands/set/
+            #    - SET see https://valkey.io/valkey-glide/python/core/#glide.async_commands.CoreCommands.set
 
-            logger.debug(f"Putting record on Redis for idempotency key: {data_record.idempotency_key}")
+            logger.debug(f"Putting record on Cache for idempotency key: {data_record.idempotency_key}")
             encoded_item = self._json_serializer(item["mapping"])
             ttl = self._get_expiry_second(expiry_timestamp=data_record.expiry_timestamp)
 
-            redis_response = self.client.set(name=data_record.idempotency_key, value=encoded_item, ex=ttl, nx=True)
+            cache_response = self.client.set(name=data_record.idempotency_key, value=encoded_item, ex=ttl, nx=True)
 
-            # If redis_response is True, the Redis SET operation was successful and the idempotency key was not
+            # If cache_response is True, the Cache SET operation was successful and the idempotency key was not
             # previously set. This indicates that we can safely proceed to the handler execution phase.
             # Most invocations should successfully proceed past this point.
-            if redis_response:
+            if cache_response:
                 return
 
-            # If redis_response is None, it indicates an existing record in Redis for the given idempotency key.
+            # If cache_response is None, it indicates an existing record in Cache for the given idempotency key.
             # This could be due to:
             # - An active idempotency record from a previous invocation that has not yet expired.
             # - An orphan record where a previous invocation has timed out.
-            # - An expired idempotency record that has not been deleted by Redis.
+            # - An expired idempotency record that has not been deleted by Cache.
             # In any case, we proceed to retrieve the record for further inspection.
 
             idempotency_record = self._get_record(data_record.idempotency_key)
@@ -431,32 +431,30 @@ def _put_in_progress_record(self, data_record: DataRecord) -> None:
 
             # Reaching this point indicates that the idempotency record found is an orphan record. An orphan record is
             # one that is neither completed nor in-progress within its expected time frame. It may result from a
-            # previous invocation that has timed out or an expired record that has yet to be cleaned up by Redis.
+            # previous invocation that has timed out or an expired record that has yet to be cleaned up by Cache.
             # We raise an error to handle this exceptional scenario appropriately.
             raise IdempotencyPersistenceConsistencyError
 
         except IdempotencyPersistenceConsistencyError:
             # Handle an orphan record by attempting to acquire a lock, which by default lasts for 10 seconds.
             # The purpose of acquiring the lock is to prevent race conditions with other processes that might
             # also be trying to handle the same orphan record. Once the lock is acquired, we set a new value
-            # for the idempotency record in Redis with the appropriate time-to-live (TTL).
+            # for the idempotency record in Cache with the appropriate time-to-live (TTL).
             with self._acquire_lock(name=item["name"]):
                 self.client.set(name=item["name"], value=encoded_item, ex=ttl)
 
             # Not removing the lock here serves as a safeguard against race conditions,
             # preventing another operation from mistakenly treating this record as an orphan while the
             # current operation is still in progress.
-        except (redis.exceptions.RedisError, redis.exceptions.RedisClusterException) as e:
-            raise e
         except Exception as e:
-            logger.debug(f"encountered non-Redis exception: {e}")
-            raise e
+            logger.debug(f"An error occurred: {e}")
+            raise
 
     @contextmanager
     def _acquire_lock(self, name: str):
         """
         Attempt to acquire a lock for a specified resource name, with a default timeout.
-        This context manager attempts to set a lock using Redis to prevent concurrent
+        This context manager attempts to set a lock using Cache to prevent concurrent
         access to a resource identified by 'name'. It uses the 'nx' flag to ensure that
         the lock is only set if it does not already exist, thereby enforcing mutual exclusion.
         """
@@ -500,15 +498,20 @@ def _update_record(self, data_record: DataRecord) -> None:
 
     def _delete_record(self, data_record: DataRecord) -> None:
         """
-        Deletes the idempotency record associated with a given DataRecord from Redis.
+        Deletes the idempotency record associated with a given DataRecord from Cache.
         This function is designed to be called after a Lambda handler invocation has completed processing.
-        It ensures that the idempotency key associated with the DataRecord is removed from Redis to
+        It ensures that the idempotency key associated with the DataRecord is removed from Cache to
         prevent future conflicts and to maintain the idempotency integrity.
 
         Note: it is essential that the idempotency key is not empty, as that would indicate the Lambda
         handler has not been invoked or the key was not properly set.
         """
         logger.debug(f"Deleting record for idempotency key: {data_record.idempotency_key}")
 
-        # See: https://redis.io/commands/del/
+        # See: https://valkey.io/valkey-glide/python/core/#glide.async_commands.CoreCommands.delete
         self.client.delete(data_record.idempotency_key)
+
+
+CachePersistenceLayer: TypeAlias = RedisCachePersistenceLayer
+CacheClientProtocol: TypeAlias = RedisClientProtocol
+CacheConnection: TypeAlias = RedisConnection