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 Set 

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 ): 

20 self._redis = None 

21 self.redis_host = redis_host 

22 self.redis_port = redis_port 

23 self.redis_db = redis_db 

24 self.processed_files: Set[str] = set() 

25 

26 self._init_cache() 

27 

28 def _init_redis(self) -> None: 

29 """Initialize Redis connection.""" 

30 try: 

31 self._redis = redis.Redis( 

32 host=self.redis_host, 

33 port=self.redis_port, 

34 db=self.redis_db, 

35 decode_responses=True, 

36 ) 

37 self._redis.ping() 

38 except RedisConnectionError: 

39 raise RuntimeError("Redis is not available. Cache requires Redis.") 

40 

41 def _init_cache(self) -> None: 

42 """Initialize cache from Redis.""" 

43 self._init_redis() 

44 existing_redis_files = self._redis.smembers(self.REDIS_KEY) 

45 self.processed_files.update(existing_redis_files) 

46 

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

48 """Add a file to the cache.""" 

49 self.processed_files.add(filename) 

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

51 

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

53 """Check if a file is in the cache.""" 

54 return filename in self.processed_files 

55 

56 def get_all(self) -> Set[str]: 

57 """Return all files in the cache.""" 

58 self.processed_files.update(self._redis.smembers(self.REDIS_KEY)) 

59 return self.processed_files