Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/site-packages/pip/_vendor/cachecontrol/caches/redis_cache.py: 43%

1# SPDX-FileCopyrightText: 2015 Eric Larson 

2# 

3# SPDX-License-Identifier: Apache-2.0 

4from __future__ import annotations 

5 

6 

7from datetime import datetime, timezone 

8from typing import TYPE_CHECKING 

9 

10from pip._vendor.cachecontrol.cache import BaseCache 

11 

12if TYPE_CHECKING: 

13 from redis import Redis 

14 

15 

16class 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