Coverage for src/lexigram/web/di/rate_limit.py: 42%

1"""Rate-limit provider for the Lexigram web integration layer.""" 

2 

3from __future__ import annotations 

4 

5from typing import TYPE_CHECKING, Any, cast 

6 

7from lexigram.contracts.core import HealthCheckResult, HealthStatus, ProviderPriority 

8from lexigram.contracts.web import WebRateLimiterProtocol 

9from lexigram.di.provider import Provider 

10from lexigram.logging import get_logger 

11from lexigram.web.middleware.rate_limit import RateLimiter 

12 

13if TYPE_CHECKING: 

14 from lexigram.contracts.core.di import ( 

15 ContainerRegistrarProtocol, 

16 ContainerResolverProtocol, 

17 ) 

18 

19logger = get_logger(__name__) 

20 

21 

22class RateLimitProvider(Provider): 

23 """Standalone provider for rate-limiting services. 

24 

25 Registers a :class:`~lexigram.web.middleware.rate_limit.RateLimiter` in the 

26 DI container and lazily resolves a Redis client from the container at boot 

27 time when none is supplied at construction. 

28 """ 

29 

30 name = "rate_limit" 

31 

32 def __init__( 

33 self, 

34 redis_client: Any = None, 

35 enabled: bool = True, 

36 ) -> None: 

37 """Initialize rate limit provider. 

38 

39 Args: 

40 redis_client: Optional pre-built Redis client for rate limiting. 

41 enabled: When ``False`` the provider skips all registration/boot. 

42 """ 

43 super().__init__(name=self.name, priority=ProviderPriority.INFRASTRUCTURE) 

44 self.redis_client = redis_client 

45 self.enabled = enabled 

46 self.limiter: RateLimiter | None = None 

47 

48 async def register(self, container: ContainerRegistrarProtocol) -> None: 

49 """Register rate limiter in the DI container. 

50 

51 Args: 

52 container: DI container registrar. 

53 """ 

54 if not self.enabled: 

55 logger.info("Rate limiting is disabled") 

56 return 

57 

58 container.singleton(WebRateLimiterProtocol, factory=lambda: self.limiter) 

59 

60 async def boot(self, container: ContainerResolverProtocol) -> None: 

61 """Start the rate limit provider, resolving Redis from the container if needed.""" 

62 if not self.enabled: 

63 return 

64 

65 if self.redis_client is None: 

66 try: 

67 self.redis_client = await cast("Any", container).resolve("redis_client") 

68 except Exception as e: # noqa: BLE001 

69 logger.warning( 

70 "Redis client not found in container (%s), rate limiting may be restricted", 

71 e, 

72 ) 

73 

74 self.limiter = RateLimiter(self.redis_client) 

75 logger.info("RateLimitProvider started") 

76 

77 async def shutdown(self) -> None: 

78 """Shutdown the rate limit provider.""" 

79 self.limiter = None 

80 logger.info("RateLimitProvider stopped") 

81 

82 async def health_check(self, timeout: float = 5.0) -> HealthCheckResult: 

83 """Check rate-limit provider health. 

84 

85 Returns: 

86 HealthCheckResult reflecting whether the limiter is initialised. 

87 """ 

88 if not self.enabled: 

89 return HealthCheckResult( 

90 component="rate_limit", 

91 status=HealthStatus.HEALTHY, 

92 details={"enabled": False}, 

93 ) 

94 

95 return HealthCheckResult( 

96 component="rate_limit", 

97 status=HealthStatus.HEALTHY if self.limiter else HealthStatus.UNHEALTHY, 

98 details={ 

99 "enabled": True, 

100 "limiter_initialized": self.limiter is not None, 

101 "has_redis_client": self.redis_client is not None, 

102 }, 

103 ) 

104 

105 

106__all__ = ["RateLimitProvider"]