Coverage for src/lexigram/web/integrations/rate_limit.py: 45%

1"""Rate limiting integration for WebProvider.""" 

2 

3from __future__ import annotations 

4 

5from typing import Any, cast 

6 

7from starlette.applications import Starlette 

8 

9from lexigram.contracts.exceptions import RateLimitError 

10from lexigram.contracts.web import WebRateLimiterProtocol 

11from lexigram.di.container import Container 

12from lexigram.logging import get_logger 

13from lexigram.web.middleware.rate_limit import ( 

14 RateLimiter, 

15 RateLimitMiddleware, 

16) 

17 

18logger = get_logger(__name__) 

19 

20 

21class RateLimitIntegration: 

22 """Handles rate limit middleware and handler configuration.""" 

23 

24 @staticmethod 

25 async def configure(app: Starlette, container: Container, web_config: Any) -> None: 

26 """Configure rate limit middleware and handlers.""" 

27 if ( 

28 not getattr(web_config, "rate_limit", None) 

29 or not web_config.rate_limit.enabled 

30 ): 

31 return 

32 

33 rate_limiter = await container.resolve_optional( 

34 cast("Any", WebRateLimiterProtocol) 

35 ) 

36 redis_client = None 

37 if rate_limiter is None: 

38 # Storage honesty: "memory" (or failed redis) constructs 

39 # RateLimiter() which logs the explicit multi-worker warning 

40 # (middleware/rate_limit.py) and then enforces in-memory. 

41 if getattr(web_config.rate_limit, "storage_backend", "memory") == "redis": 

42 try: 

43 redis_client = await container.resolve("redis_client") 

44 except Exception as redis_err: # noqa: BLE001 — degrade to memory 

45 logger.warning( 

46 "redis_client unresolvable; using in-memory: %r", redis_err 

47 ) 

48 rate_limiter = RateLimiter(redis_client) 

49 

50 if rate_limiter: 

51 app.add_exception_handler( 

52 RateLimitError, 

53 RateLimitIntegration._rate_limit_handler, 

54 ) 

55 app.add_middleware( 

56 RateLimitMiddleware, 

57 rate_limiter=cast("Any", rate_limiter), 

58 config=cast("Any", web_config.rate_limit), 

59 ) 

60 logger.info("Rate limiting middleware configured") 

61 

62 @staticmethod 

63 async def _rate_limit_handler(_request: Any, exc: Exception) -> Any: 

64 """Standard rate limit exceeded handler.""" 

65 from lexigram.web.middleware.rate_limit import _rate_limit_429_response 

66 

67 return _rate_limit_429_response(exc)