Coverage for src\utility\rediss.py: 100%

1from collections.abc import AsyncIterator 

2from contextlib import asynccontextmanager 

3from fastapi import FastAPI 

4from fastapi_cache import FastAPICache 

5from fastapi_cache.backends.redis import RedisBackend 

6from fastapi_cache.decorator import cache #noqa: F401 

7from fastapi_cache.types import Backend 

8from redis import asyncio as aioredis 

9from redis.exceptions import ConnectionError 

10from src.utility.config import Config 

11from src.utility.lib import Logger 

12from typing import Optional, Tuple, override 

13 

14 

15class NoOpBackend(Backend): 

16 @override 

17 async def get(self, key: str) -> Optional[bytes]: # pragma: no cover 

18 return None 

19 

20 @override 

21 async def set(self, key: str, value: bytes, expire: Optional[int] = None) -> None: # pragma: no cover 

22 pass 

23 

24 @override 

25 async def clear(self, namespace: Optional[str] = None, key: Optional[str] = None) -> int: # pragma: no cover 

26 return 0 

27 

28 @override 

29 async def get_with_ttl(self, key: str) -> Tuple[int, Optional[bytes]]: # pragma: no cover 

30 return 0, None 

31 

32 

33class SafeRedisBackend(RedisBackend): 

34 @override 

35 async def get_with_ttl(self, key: str) -> Tuple[int, Optional[bytes]]: # pragma: no cover 

36 async with self.redis.pipeline(transaction=False) as pipe: # transaction is set to False to prevent this Redis error: "Error in get_with_ttl: unknown command 'EXEC'" 

37 return await pipe.ttl(key).get(key).execute() 

38 

39 

40def lifespan_factory(enable_cache: bool, flush_cache: bool): 

41 @asynccontextmanager 

42 async def lifespan(_: FastAPI) -> AsyncIterator[None]: 

43 FastAPICache._init = False 

44 

45 if not enable_cache: 

46 FastAPICache.init(backend=NoOpBackend()) 

47 Logger.warn("Redis caching disabled - configuration.") 

48 

49 yield 

50 return 

51 

52 try: 

53 redis = aioredis.from_url( 

54 url=Config.REDIS_SCHEME, 

55 host=Config.REDIS_HOST, 

56 port=Config.REDIS_PORT, 

57 username=Config.REDIS_USERNAME, 

58 password=Config.REDIS_PASSWORD 

59 ) 

60 await redis.ping() 

61 

62 if flush_cache: 

63 await redis.flushall() 

64 Logger.warn("Redis cache flushed on startup.") 

65 

66 FastAPICache.init(backend=SafeRedisBackend(redis), prefix="fastapi-cache") 

67 Logger.info("Redis caching enabled.") 

68 

69 except ConnectionError: 

70 FastAPICache.init(backend=NoOpBackend()) 

71 Logger.exception("Redis caching disabled - connection issue.") 

72 

73 except Exception: 

74 FastAPICache.init(backend=NoOpBackend()) 

75 Logger.exception("Redis caching disabled - unexpected error.") 

76 

77 yield 

78 

79 return lifespan