Coverage for src/lexigram/web/integrations/throttle.py: 33%

1"""@throttle decorator and RateLimitModule — ergonomic rate limiting sugar. 

2 

3Provides a NestJS-style ``@throttle("30/minute")`` decorator that sits on top 

4of the lower-level :func:`~lexigram.web.middleware.rate_limit.rate_limit` 

5decorator, plus a ``RateLimitModule`` for module-level configuration. 

6 

7Usage:: 

8 

9 from lexigram.web import throttle, RateLimitModule 

10 

11 # Boot-time module config 

12 app.add_module(RateLimitModule.configure( 

13 backend="memory", # "redis" | "memory" 

14 default_limit="100/minute", # applied when no route-level @throttle 

15 )) 

16 

17 class APIController(Controller): 

18 @get("/search") 

19 @throttle("30/minute") 

20 async def search(self, request: Request) -> list[dict]: ... 

21 

22 @post("/upload") 

23 @throttle("5/hour", by="user") 

24 async def upload(self, request: Request) -> dict: ... 

25 

26 @get("/public") 

27 @throttle("1000/hour", by="ip") 

28 async def public_data(self, request: Request) -> dict: ... 

29""" 

30 

31from __future__ import annotations 

32 

33import re 

34from typing import TYPE_CHECKING, Any, Literal 

35 

36if TYPE_CHECKING: 

37 from collections.abc import Callable 

38 

39__all__ = ["RateLimitModule", "throttle"] 

40 

41# --------------------------------------------------------------------------- 

42# Rate string parsing 

43# --------------------------------------------------------------------------- 

44 

45_WINDOW_MAP: dict[str, int] = { 

46 "second": 1, 

47 "seconds": 1, 

48 "minute": 60, 

49 "minutes": 60, 

50 "hour": 3600, 

51 "hours": 3600, 

52 "day": 86400, 

53 "days": 86400, 

54} 

55 

56_RATE_RE = re.compile( 

57 r"^(\d+)\s*/\s*(" + "|".join(_WINDOW_MAP) + r")$", 

58 re.IGNORECASE, 

59) 

60 

61 

62def _parse_rate(rate: str) -> tuple[int, int]: 

63 """Parse a rate string like ``"30/minute"`` into ``(max_requests, window_seconds)``. 

64 

65 Args: 

66 rate: Rate string in the format ``"<count>/<unit>"`` where unit is 

67 one of ``second``, ``minute``, ``hour``, or ``day`` (plural forms 

68 are also accepted). 

69 

70 Returns: 

71 Tuple of ``(max_requests, window_seconds)``. 

72 

73 Raises: 

74 ValueError: If the rate string is not in the expected format. 

75 """ 

76 m = _RATE_RE.match(rate.strip()) 

77 if not m: 

78 raise ValueError( 

79 f"Invalid rate string {rate!r}. " 

80 "Expected format: '<count>/<unit>' where unit is " 

81 "second/minute/hour/day (e.g. '30/minute', '5/hour')." 

82 ) 

83 max_requests = int(m.group(1)) 

84 window_seconds = _WINDOW_MAP[m.group(2).lower()] 

85 return max_requests, window_seconds 

86 

87 

88# --------------------------------------------------------------------------- 

89# @throttle decorator 

90# --------------------------------------------------------------------------- 

91 

92 

93def throttle( 

94 rate: str, 

95 *, 

96 by: Literal["user", "ip", "endpoint"] = "user", 

97) -> Callable[[Any], Any]: 

98 """Rate-limit a route handler using a human-readable rate string. 

99 

100 This is ergonomic sugar over the lower-level 

101 :func:`~lexigram.web.middleware.rate_limit.rate_limit` decorator. 

102 

103 Args: 

104 rate: Rate string in ``"<count>/<unit>"`` format, e.g. 

105 ``"30/minute"``, ``"5/hour"``, ``"100/second"``. 

106 by: Scope for the rate limit key: 

107 

108 * ``"user"`` — per authenticated user (falls back to IP when 

109 ``request.state.user`` is not set). 

110 * ``"ip"`` — per client IP address. 

111 * ``"endpoint"`` — per route path + HTTP method. 

112 

113 Returns: 

114 Decorator that enforces the rate limit on the decorated handler. 

115 

116 Raises: 

117 ValueError: If *rate* is not a valid rate string. 

118 

119 Example:: 

120 

121 @get("/search") 

122 @throttle("30/minute") 

123 async def search(self, request: Request) -> list[dict]: ... 

124 

125 @post("/upload") 

126 @throttle("5/hour", by="user") 

127 async def upload(self, request: Request) -> dict: ... 

128 """ 

129 max_requests, window_seconds = _parse_rate(rate) 

130 

131 from lexigram.web.middleware.rate_limit import rate_limit as _rate_limit 

132 

133 return _rate_limit( 

134 max_requests=max_requests, 

135 window_seconds=window_seconds, 

136 scope=by, 

137 ) 

138 

139 

140# --------------------------------------------------------------------------- 

141# RateLimitModule 

142# --------------------------------------------------------------------------- 

143 

144 

145class RateLimitModule: 

146 """Module-level rate limiting configuration. 

147 

148 Registers a :class:`~lexigram.web.di.rate_limit.RateLimitProvider` in 

149 the application's DI container so that route-level ``@throttle`` and 

150 ``@rate_limit`` decorators can resolve a limiter from the container. 

151 

152 Usage:: 

153 

154 app.add_module(RateLimitModule.configure( 

155 backend="redis", 

156 default_limit="100/minute", 

157 )) 

158 

159 # Development / testing — no external service required: 

160 app.add_module(RateLimitModule.configure(backend="memory")) 

161 """ 

162 

163 @classmethod 

164 def configure( 

165 cls, 

166 *, 

167 backend: Literal["redis", "memory"] = "memory", 

168 default_limit: str | None = None, 

169 redis_client: Any = None, 

170 enabled: bool = True, 

171 ) -> Any: 

172 """Create a :class:`~lexigram.di.module.DynamicModule` that wires up 

173 rate limiting. 

174 

175 Args: 

176 backend: Storage backend. ``"redis"`` uses atomic Lua scripts 

177 (recommended for production); ``"memory``" uses a 

178 process-local sliding window (suitable for development and 

179 single-process deployments). 

180 default_limit: Optional default rate limit applied globally (not 

181 yet enforced by the module — provided for future middleware 

182 use). Accepts the same rate string format as 

183 :func:`throttle`. 

184 redis_client: Pre-built Redis client. Only used when 

185 ``backend="redis"``. When *None*, the provider tries to 

186 resolve a ``"redis_client"`` binding from the container at 

187 boot time. 

188 enabled: When ``False`` the rate limiter is not started. 

189 

190 Returns: 

191 A :class:`~lexigram.di.module.DynamicModule` descriptor. 

192 """ 

193 from lexigram.di.module import DynamicModule 

194 from lexigram.web.di.rate_limit import RateLimitProvider 

195 

196 if default_limit is not None: 

197 # Validate at module-configuration time so errors surface early. 

198 _parse_rate(default_limit) 

199 

200 provider_kwargs: dict[str, Any] = {"enabled": enabled} 

201 if backend == "redis" and redis_client is not None: 

202 provider_kwargs["redis_client"] = redis_client 

203 

204 rate_provider = RateLimitProvider(**provider_kwargs) 

205 

206 # Build a minimal "wrapper" module class to satisfy DynamicModule 

207 from lexigram.di.module import module 

208 

209 @module() 

210 class _RateLimitModuleClass: 

211 pass 

212 

213 return DynamicModule( 

214 module=_RateLimitModuleClass, 

215 providers=[rate_provider], 

216 is_global=True, 

217 )