Coverage for src / piccione / upload / cache_manager.py: 100%

1# SPDX-FileCopyrightText: 2025 Arcangelo Massari <arcangelo.massari@unibo.it> 

2# 

3# SPDX-License-Identifier: ISC 

4 

5from typing import cast 

6 

7import redis 

8from redis.exceptions import ConnectionError as RedisConnectionError 

9 

10 

11class CacheManager: 

12 REDIS_KEY = "processed_files" 

13 

14 def __init__( 

15 self, 

16 redis_host: str = "localhost", 

17 redis_port: int = 6379, 

18 redis_db: int = 4, 

19 ) -> None: 

20 self.processed_files: set[str] = set() 

21 try: 

22 self._redis = redis.Redis( 

23 host=redis_host, 

24 port=redis_port, 

25 db=redis_db, 

26 decode_responses=True, 

27 ) 

28 self._redis.ping() 

29 except RedisConnectionError as err: 

30 msg = "Redis is not available. Cache requires Redis." 

31 raise RuntimeError(msg) from err 

32 self.processed_files.update(cast("set[str]", self._redis.smembers(self.REDIS_KEY))) 

33 

34 def add(self, filename: str) -> None: 

35 self.processed_files.add(filename) 

36 self._redis.sadd(self.REDIS_KEY, filename) 

37 

38 def __contains__(self, filename: str) -> bool: 

39 return filename in self.processed_files 

40 

41 def get_all(self) -> set[str]: 

42 self.processed_files.update(cast("set[str]", self._redis.smembers(self.REDIS_KEY))) 

43 return self.processed_files