Coverage for polar/redis.py: 90%

1from typing import TYPE_CHECKING, Literal, TypeAlias 

2 

3import redis.asyncio as _async_redis 

4from fastapi import Request 

5from redis import ConnectionError, RedisError, TimeoutError 

6from redis.asyncio.retry import Retry 

7from redis.backoff import default_backoff 

8 

9from polar.config import settings 

10 

11# https://github.com/python/typeshed/issues/7597#issuecomment-1117551641 

12# Redis is generic at type checking, but not at runtime... 

13if TYPE_CHECKING: 13 ↛ 14line 13 didn't jump to line 14 because the condition on line 13 was never true

14 Redis = _async_redis.Redis[str] 

15else: 

16 Redis = _async_redis.Redis 

17 

18 

19REDIS_RETRY_ON_ERRROR: list[type[RedisError]] = [ConnectionError, TimeoutError] 

20REDIS_RETRY = Retry(default_backoff(), retries=50) 

21 

22ProcessName: TypeAlias = Literal["app", "rate-limit", "worker", "script"] 

23 

24 

25def create_redis(process_name: ProcessName) -> Redis: 

26 return _async_redis.Redis.from_url( 

27 settings.redis_url, 

28 decode_responses=True, 

29 retry_on_error=REDIS_RETRY_ON_ERRROR, 

30 retry=REDIS_RETRY, 

31 client_name=f"{settings.ENV.value}.{process_name}", 

32 ) 

33 

34 

35async def get_redis(request: Request) -> Redis: 

36 return request.state.redis 

37 

38 

39__all__ = [ 

40 "Redis", 

41 "REDIS_RETRY_ON_ERRROR", 

42 "REDIS_RETRY", 

43 "create_redis", 

44 "get_redis", 

45]