Coverage for src/lexigram/graphql/security/rate_limit.py: 78%

1"""Rate limiting for GraphQL. 

2 

3This module provides rate limiting for GraphQL operations. 

4""" 

5 

6from __future__ import annotations 

7 

8from dataclasses import dataclass 

9from typing import Any, ClassVar 

10 

11from lexigram.config.base import BaseConfig 

12from lexigram.contracts.exceptions import RateLimitError as ContractsRateLimitError 

13from lexigram.contracts.web import WebRateLimiterProtocol 

14from lexigram.validation import ConfigDict, Field 

15 

16 

17@dataclass(init=False) 

18class RateLimitConfig(BaseConfig): 

19 """Configuration for rate limiting. 

20 

21 Attributes: 

22 enabled: Whether rate limiting is enabled. 

23 requests_per_minute: Number of allowed requests per minute. 

24 burst_limit: Maximum burst size. 

25 """ 

26 

27 model_config: ClassVar[ConfigDict] = ConfigDict(extra="ignore") 

28 

29 enabled: bool = True 

30 requests_per_minute: int = Field(default=60, ge=1) 

31 burst_limit: int = Field(default=10, ge=1) 

32 

33 

34class RateLimiter: 

35 """Base class for rate limiters.""" 

36 

37 async def is_allowed(self, context: Any, **kwargs: Any) -> bool: 

38 """Check if a request is allowed.""" 

39 return True 

40 

41 

42class UnifiedRateLimiter(RateLimiter): 

43 """Rate limiter that delegates to an injected web rate limiter.""" 

44 

45 def __init__(self, web_rate_limiter: WebRateLimiterProtocol | None = None) -> None: 

46 self._web_rate_limiter = web_rate_limiter 

47 

48 async def is_allowed(self, context: Any, **kwargs: Any) -> bool: 

49 """Check if a request is allowed using the web rate limiter.""" 

50 # Unpack kwargs 

51 max_requests = kwargs.get("max_requests", 60) 

52 window_seconds = kwargs.get("window_seconds", 60) 

53 scope = kwargs.get("scope", "user") 

54 

55 if not hasattr(context, "raw_request") or not context.raw_request: 

56 return True 

57 

58 if self._web_rate_limiter is None: 

59 return True 

60 

61 try: 

62 await self._web_rate_limiter.check_rate_limit( 

63 context.raw_request, 

64 max_requests=max_requests, 

65 window_seconds=window_seconds, 

66 scope=scope, 

67 ) 

68 return True 

69 except ContractsRateLimitError as exc: 

70 from lexigram.graphql.exceptions import RateLimitError 

71 

72 raise RateLimitError("Rate limit exceeded") from exc 

73 except (LookupError, RuntimeError, AttributeError, TypeError, ValueError): 

74 # On infrastructure errors (Redis down, etc.) allow the request. 

75 return True 

76 

77 

78__all__ = [ 

79 "RateLimitConfig", 

80 "RateLimiter", 

81 "UnifiedRateLimiter", 

82]