| 14 | |
| 15 | |
| 16 | class RedisCache(BaseCache): |
| 17 | def __init__(self, conn: Redis[bytes]) -> None: |
| 18 | self.conn = conn |
| 19 | |
| 20 | def get(self, key: str) -> bytes | None: |
| 21 | return self.conn.get(key) |
| 22 | |
| 23 | def set( |
| 24 | self, key: str, value: bytes, expires: int | datetime | None = None |
| 25 | ) -> None: |
| 26 | if not expires: |
| 27 | self.conn.set(key, value) |
| 28 | elif isinstance(expires, datetime): |
| 29 | now_utc = datetime.now(timezone.utc) |
| 30 | if expires.tzinfo is None: |
| 31 | now_utc = now_utc.replace(tzinfo=None) |
| 32 | delta = expires - now_utc |
| 33 | self.conn.setex(key, int(delta.total_seconds()), value) |
| 34 | else: |
| 35 | self.conn.setex(key, expires, value) |
| 36 | |
| 37 | def delete(self, key: str) -> None: |
| 38 | self.conn.delete(key) |
| 39 | |
| 40 | def clear(self) -> None: |
| 41 | """Helper for clearing all the keys in a database. Use with |
| 42 | caution!""" |
| 43 | for key in self.conn.keys(): |
| 44 | self.conn.delete(key) |
| 45 | |
| 46 | def close(self) -> None: |
| 47 | """Redis uses connection pooling, no need to close the connection.""" |
| 48 | pass |
no outgoing calls
searching dependent graphs…