0
# Rate Limiting Strategies
1
2
Rate limiting strategies implement different algorithms for enforcing rate limits, each with distinct characteristics for accuracy, memory usage, and computational efficiency. The choice of strategy depends on your specific requirements for precision, resource usage, and acceptable trade-offs.
3
4
## Capabilities
5
6
### Base Rate Limiter
7
8
Abstract base class defining the common interface for all rate limiting strategies.
9
10
```python { .api }
11
from abc import ABC, abstractmethod
12
from limits.storage import Storage
13
from limits.limits import RateLimitItem
14
from limits.util import WindowStats
15
16
class RateLimiter(ABC):
17
"""Abstract base class for rate limiting strategies"""
18
19
def __init__(self, storage: Storage):
20
"""
21
Initialize rate limiter with storage backend.
22
23
Args:
24
storage: Storage backend for persisting rate limit data
25
"""
26
27
@abstractmethod
28
def hit(self, item: RateLimitItem, *identifiers: str, cost: int = 1) -> bool:
29
"""
30
Consume the rate limit and return whether request is allowed.
31
32
Args:
33
item: The rate limit item defining limits
34
identifiers: Unique identifiers for this limit instance
35
cost: Cost of this request (default: 1)
36
37
Returns:
38
True if request is allowed, False if rate limit exceeded
39
"""
40
41
@abstractmethod
42
def test(self, item: RateLimitItem, *identifiers: str, cost: int = 1) -> bool:
43
"""
44
Check if rate limit allows request without consuming it.
45
46
Args:
47
item: The rate limit item defining limits
48
identifiers: Unique identifiers for this limit instance
49
cost: Expected cost to be consumed (default: 1)
50
51
Returns:
52
True if request would be allowed, False otherwise
53
"""
54
55
@abstractmethod
56
def get_window_stats(self, item: RateLimitItem, *identifiers: str) -> WindowStats:
57
"""
58
Get current window statistics.
59
60
Args:
61
item: The rate limit item defining limits
62
identifiers: Unique identifiers for this limit instance
63
64
Returns:
65
WindowStats with reset_time and remaining quota
66
"""
67
68
def clear(self, item: RateLimitItem, *identifiers: str) -> None:
69
"""
70
Clear rate limit data for the given identifiers.
71
72
Args:
73
item: The rate limit item defining limits
74
identifiers: Unique identifiers for this limit instance
75
"""
76
```
77
78
### Fixed Window Strategy
79
80
Divides time into fixed windows and tracks request counts per window. Simple and memory-efficient but allows bursts at window boundaries.
81
82
```python { .api }
83
class FixedWindowRateLimiter(RateLimiter):
84
"""
85
Fixed window rate limiting strategy.
86
87
Divides time into fixed windows (e.g., each minute from :00 to :59).
88
Allows up to the limit within each window. Memory efficient but can
89
allow bursts of up to 2x the limit at window boundaries.
90
"""
91
92
def hit(self, item: RateLimitItem, *identifiers: str, cost: int = 1) -> bool:
93
"""
94
Consume from the rate limit.
95
96
Increments the counter for the current fixed window. If the
97
increment would exceed the limit, the request is rejected.
98
99
Args:
100
item: Rate limit configuration
101
identifiers: Unique identifiers (user ID, IP, etc.)
102
cost: Cost of this request (default: 1)
103
104
Returns:
105
True if request allowed, False if limit exceeded
106
"""
107
108
def test(self, item: RateLimitItem, *identifiers: str, cost: int = 1) -> bool:
109
"""
110
Check if request would be allowed without consuming.
111
112
Args:
113
item: Rate limit configuration
114
identifiers: Unique identifiers
115
cost: Expected cost (default: 1)
116
117
Returns:
118
True if request would be allowed
119
"""
120
121
def get_window_stats(self, item: RateLimitItem, *identifiers: str) -> WindowStats:
122
"""
123
Get current window statistics.
124
125
Returns:
126
WindowStats(reset_time, remaining) where reset_time is when
127
the current fixed window expires
128
"""
129
```
130
131
### Moving Window Strategy
132
133
Maintains a sliding window of individual request timestamps, providing the most accurate rate limiting but with higher memory usage.
134
135
```python { .api }
136
class MovingWindowRateLimiter(RateLimiter):
137
"""
138
Moving window rate limiting strategy.
139
140
Maintains individual timestamps for each request within the time window.
141
Provides the most accurate rate limiting by checking requests within
142
exactly the specified time period, but uses more memory.
143
144
Requires storage backend with MovingWindowSupport.
145
"""
146
147
def __init__(self, storage: Storage):
148
"""
149
Initialize moving window rate limiter.
150
151
Args:
152
storage: Storage backend that supports acquire_entry and
153
get_moving_window methods
154
155
Raises:
156
NotImplementedError: If storage doesn't support moving window
157
"""
158
159
def hit(self, item: RateLimitItem, *identifiers: str, cost: int = 1) -> bool:
160
"""
161
Consume from the rate limit using moving window.
162
163
Adds timestamp entries for the request and removes expired ones.
164
Checks if total requests in the sliding window exceed the limit.
165
166
Args:
167
item: Rate limit configuration
168
identifiers: Unique identifiers
169
cost: Cost of this request (default: 1)
170
171
Returns:
172
True if request allowed, False if limit exceeded
173
"""
174
175
def test(self, item: RateLimitItem, *identifiers: str, cost: int = 1) -> bool:
176
"""
177
Check if request would be allowed in moving window.
178
179
Args:
180
item: Rate limit configuration
181
identifiers: Unique identifiers
182
cost: Expected cost (default: 1)
183
184
Returns:
185
True if request would be allowed
186
"""
187
188
def get_window_stats(self, item: RateLimitItem, *identifiers: str) -> WindowStats:
189
"""
190
Get moving window statistics.
191
192
Returns:
193
WindowStats(reset_time, remaining) where reset_time is when
194
the oldest request in the window expires
195
"""
196
```
197
198
### Sliding Window Counter Strategy
199
200
Approximates a moving window using weighted counters from current and previous fixed windows. Balances accuracy and memory efficiency.
201
202
```python { .api }
203
class SlidingWindowCounterRateLimiter(RateLimiter):
204
"""
205
Sliding window counter rate limiting strategy.
206
207
Approximates a moving window by weighting the current and previous
208
fixed window counters based on time elapsed. More memory efficient
209
than moving window while more accurate than fixed window.
210
211
Added in version 4.1. Requires storage with SlidingWindowCounterSupport.
212
"""
213
214
def __init__(self, storage: Storage):
215
"""
216
Initialize sliding window counter rate limiter.
217
218
Args:
219
storage: Storage backend supporting get_sliding_window and
220
acquire_sliding_window_entry methods
221
222
Raises:
223
NotImplementedError: If storage doesn't support sliding window counter
224
"""
225
226
def hit(self, item: RateLimitItem, *identifiers: str, cost: int = 1) -> bool:
227
"""
228
Consume from rate limit using sliding window counter.
229
230
Uses weighted combination of current and previous window counters
231
to approximate the moving window behavior.
232
233
Args:
234
item: Rate limit configuration
235
identifiers: Unique identifiers
236
cost: Cost of this request (default: 1)
237
238
Returns:
239
True if request allowed, False if limit exceeded
240
"""
241
242
def test(self, item: RateLimitItem, *identifiers: str, cost: int = 1) -> bool:
243
"""
244
Check if request would be allowed using sliding window counter.
245
246
Args:
247
item: Rate limit configuration
248
identifiers: Unique identifiers
249
cost: Expected cost (default: 1)
250
251
Returns:
252
True if request would be allowed
253
"""
254
255
def get_window_stats(self, item: RateLimitItem, *identifiers: str) -> WindowStats:
256
"""
257
Get sliding window counter statistics.
258
259
Returns:
260
WindowStats(reset_time, remaining) with approximated values
261
based on weighted counters
262
"""
263
```
264
265
### Legacy Fixed Window with Elastic Expiry (Deprecated)
266
267
Fixed window strategy with elastic expiry behavior. Deprecated in version 4.1.
268
269
```python { .api }
270
class FixedWindowElasticExpiryRateLimiter(FixedWindowRateLimiter):
271
"""
272
Fixed window with elastic expiry rate limiter.
273
274
Similar to FixedWindowRateLimiter but with elastic expiry behavior.
275
276
Deprecated in version 4.1 - use FixedWindowRateLimiter instead.
277
"""
278
279
def hit(self, item: RateLimitItem, *identifiers: str, cost: int = 1) -> bool:
280
"""
281
Consume from rate limit with elastic expiry.
282
283
Args:
284
item: Rate limit configuration
285
identifiers: Unique identifiers
286
cost: Cost of this request (default: 1)
287
288
Returns:
289
True if request allowed, False if limit exceeded
290
"""
291
```
292
293
## Strategy Selection
294
295
```python { .api }
296
# Type alias for all known strategy types
297
KnownStrategy = (
298
type[SlidingWindowCounterRateLimiter]
299
| type[FixedWindowRateLimiter]
300
| type[FixedWindowElasticExpiryRateLimiter]
301
| type[MovingWindowRateLimiter]
302
)
303
304
# Strategy registry mapping names to classes
305
STRATEGIES: dict[str, KnownStrategy] = {
306
"fixed-window": FixedWindowRateLimiter,
307
"moving-window": MovingWindowRateLimiter,
308
"sliding-window-counter": SlidingWindowCounterRateLimiter,
309
"fixed-window-elastic-expiry": FixedWindowElasticExpiryRateLimiter
310
}
311
```
312
313
## Usage Examples
314
315
### Basic Strategy Usage
316
317
```python
318
from limits import RateLimitItemPerMinute
319
from limits.storage import storage_from_string
320
from limits.strategies import (
321
FixedWindowRateLimiter,
322
MovingWindowRateLimiter,
323
SlidingWindowCounterRateLimiter
324
)
325
326
# Set up rate limit and storage
327
rate_limit = RateLimitItemPerMinute(100) # 100 requests per minute
328
storage = storage_from_string("redis://localhost:6379")
329
330
# Choose strategy based on requirements
331
fixed_limiter = FixedWindowRateLimiter(storage) # Memory efficient
332
moving_limiter = MovingWindowRateLimiter(storage) # Most accurate
333
sliding_limiter = SlidingWindowCounterRateLimiter(storage) # Balanced
334
335
user_id = "user123"
336
337
# Test and consume with any strategy
338
if fixed_limiter.test(rate_limit, user_id):
339
success = fixed_limiter.hit(rate_limit, user_id)
340
if success:
341
print("Request allowed")
342
else:
343
print("Rate limit exceeded")
344
345
# Get window statistics
346
stats = fixed_limiter.get_window_stats(rate_limit, user_id)
347
print(f"Remaining: {stats.remaining}")
348
print(f"Reset time: {stats.reset_time}")
349
```
350
351
### Strategy Comparison
352
353
```python
354
import time
355
from limits import RateLimitItemPerMinute
356
from limits.storage import MemoryStorage
357
from limits.strategies import (
358
FixedWindowRateLimiter,
359
MovingWindowRateLimiter,
360
SlidingWindowCounterRateLimiter
361
)
362
363
# Create rate limit and storage
364
rate_limit = RateLimitItemPerMinute(10) # 10 requests per minute
365
storage = MemoryStorage()
366
367
# Initialize all strategies
368
fixed = FixedWindowRateLimiter(storage)
369
moving = MovingWindowRateLimiter(storage)
370
sliding = SlidingWindowCounterRateLimiter(storage)
371
372
user_id = "test_user"
373
374
# Fixed Window: May allow bursts at window boundaries
375
for i in range(15):
376
result = fixed.hit(rate_limit, f"{user_id}_fixed", i)
377
print(f"Fixed window request {i}: {'allowed' if result else 'denied'}")
378
379
# Moving Window: Precise tracking of each request timestamp
380
for i in range(15):
381
result = moving.hit(rate_limit, f"{user_id}_moving", i)
382
print(f"Moving window request {i}: {'allowed' if result else 'denied'}")
383
time.sleep(0.1) # Small delay between requests
384
385
# Sliding Window Counter: Approximation using weighted counters
386
for i in range(15):
387
result = sliding.hit(rate_limit, f"{user_id}_sliding", i)
388
print(f"Sliding window request {i}: {'allowed' if result else 'denied'}")
389
```
390
391
### Working with Costs
392
393
```python
394
from limits import RateLimitItemPerSecond
395
from limits.storage import MemoryStorage
396
from limits.strategies import FixedWindowRateLimiter
397
398
# Rate limit allowing 100 "cost units" per second
399
rate_limit = RateLimitItemPerSecond(100)
400
storage = MemoryStorage()
401
limiter = FixedWindowRateLimiter(storage)
402
403
api_key = "api_key_123"
404
405
# Different operations have different costs
406
light_request_cost = 1 # GET requests
407
heavy_request_cost = 10 # POST requests with file upload
408
batch_request_cost = 50 # Batch operations
409
410
# Check if expensive operation is allowed
411
if limiter.test(rate_limit, api_key, cost=heavy_request_cost):
412
success = limiter.hit(rate_limit, api_key, cost=heavy_request_cost)
413
if success:
414
print("Heavy request processed")
415
416
# Check remaining quota
417
stats = limiter.get_window_stats(rate_limit, api_key)
418
print(f"Remaining cost units: {stats.remaining}")
419
```