Coverage for polar/redis.py: 90%

1from typing import TYPE_CHECKING, Literal, TypeAlias 2 ctx1ab

2 

3import redis.asyncio as _async_redis 2 ctx1ab

4from fastapi import Request 2 ctx1ab

5from redis import ConnectionError, RedisError, TimeoutError 2 ctx1ab

6from redis.asyncio.retry import Retry 2 ctx1ab

7from redis.backoff import default_backoff 2 ctx1ab

8 

9from polar.config import settings 2 ctx1ab

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 true2 ctx1ab

14 Redis = _async_redis.Redis[str] 

15else: 

16 Redis = _async_redis.Redis 2 ctx1ab

17 

18 

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

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

21 

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

23 

24 

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

26 return _async_redis.Redis.from_url( (empty)

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: 2 ctx1ab

36 return request.state.redis (empty)

37 

38 

39__all__ = [ 2 ctx1ab

40 "Redis", 

41 "REDIS_RETRY_ON_ERRROR", 

42 "REDIS_RETRY", 

43 "create_redis", 

44 "get_redis", 

45]