about summary refs log tree commit diff
path: root/.venv/lib/python3.12/site-packages/pip/_vendor/cachecontrol/caches/redis_cache.py
diff options
context:
space:
mode:
Diffstat (limited to '.venv/lib/python3.12/site-packages/pip/_vendor/cachecontrol/caches/redis_cache.py')
-rw-r--r--.venv/lib/python3.12/site-packages/pip/_vendor/cachecontrol/caches/redis_cache.py48
1 files changed, 48 insertions, 0 deletions
diff --git a/.venv/lib/python3.12/site-packages/pip/_vendor/cachecontrol/caches/redis_cache.py b/.venv/lib/python3.12/site-packages/pip/_vendor/cachecontrol/caches/redis_cache.py
new file mode 100644
index 00000000..f4f68c47
--- /dev/null
+++ b/.venv/lib/python3.12/site-packages/pip/_vendor/cachecontrol/caches/redis_cache.py
@@ -0,0 +1,48 @@
+# SPDX-FileCopyrightText: 2015 Eric Larson
+#
+# SPDX-License-Identifier: Apache-2.0
+from __future__ import annotations
+
+
+from datetime import datetime, timezone
+from typing import TYPE_CHECKING
+
+from pip._vendor.cachecontrol.cache import BaseCache
+
+if TYPE_CHECKING:
+    from redis import Redis
+
+
+class RedisCache(BaseCache):
+    def __init__(self, conn: Redis[bytes]) -> None:
+        self.conn = conn
+
+    def get(self, key: str) -> bytes | None:
+        return self.conn.get(key)
+
+    def set(
+        self, key: str, value: bytes, expires: int | datetime | None = None
+    ) -> None:
+        if not expires:
+            self.conn.set(key, value)
+        elif isinstance(expires, datetime):
+            now_utc = datetime.now(timezone.utc)
+            if expires.tzinfo is None:
+                now_utc = now_utc.replace(tzinfo=None)
+            delta = expires - now_utc
+            self.conn.setex(key, int(delta.total_seconds()), value)
+        else:
+            self.conn.setex(key, expires, value)
+
+    def delete(self, key: str) -> None:
+        self.conn.delete(key)
+
+    def clear(self) -> None:
+        """Helper for clearing all the keys in a database. Use with
+        caution!"""
+        for key in self.conn.keys():
+            self.conn.delete(key)
+
+    def close(self) -> None:
+        """Redis uses connection pooling, no need to close the connection."""
+        pass