0
# Utilities
1

2
Helper functions and classes for parsing rate limit strings, managing window statistics, handling dependencies, and working with the limits library ecosystem.
3

4
## Capabilities
5

6
### Rate Limit Parsing
7

8
Functions for converting human-readable rate limit strings into RateLimitItem objects, supporting both single and multiple rate limit specifications.
9

10
```python { .api }
11
def parse(limit_string: str) -> RateLimitItem:
12
    """
13
    Parse a single rate limit string into a RateLimitItem.
14
    
15
    Converts human-readable rate limit notation into the appropriate
16
    RateLimitItem subclass based on the granularity specified.
17
    
18
    Args:
19
        limit_string: Rate limit string like "10/second", "5 per minute", 
20
                     "100/hour", "1000 per day"
21
    
22
    Returns:
23
        RateLimitItem instance matching the specified granularity
24
        
25
    Raises:
26
        ValueError: If the string notation is invalid or unparseable
27
        
28
    Examples:
29
        parse("10/second")     # Returns RateLimitItemPerSecond(10)
30
        parse("5 per minute")  # Returns RateLimitItemPerMinute(5)
31
        parse("100/hour")      # Returns RateLimitItemPerHour(100)
32
        parse("1000 per day")  # Returns RateLimitItemPerDay(1000)
33
    """
34

35
def parse_many(limit_string: str) -> list[RateLimitItem]:
36
    """
37
    Parse multiple rate limit strings separated by delimiters.
38
    
39
    Supports comma, semicolon, or pipe-separated rate limit specifications,
40
    allowing complex multi-tier rate limiting configurations in a single string.
41
    
42
    Args:
43
        limit_string: Multiple rate limits like "10/second; 100/minute; 1000/hour"
44
        
45
    Returns:
46
        List of RateLimitItem instances for each parsed rate limit
47
        
48
    Raises:
49
        ValueError: If any part of the string notation is invalid
50
        
51
    Examples:
52
        parse_many("10/second; 100/minute")
53
        # Returns [RateLimitItemPerSecond(10), RateLimitItemPerMinute(100)]
54
        
55
        parse_many("5/second, 50/minute, 500/hour")  
56
        # Returns [RateLimitItemPerSecond(5), RateLimitItemPerMinute(50), RateLimitItemPerHour(500)]
57
        
58
        parse_many("1 per second | 10 per minute")
59
        # Returns [RateLimitItemPerSecond(1), RateLimitItemPerMinute(10)]
60
    """
61

62
def granularity_from_string(granularity_string: str) -> type[RateLimitItem]:
63
    """
64
    Get RateLimitItem class for a granularity string.
65
    
66
    Maps granularity names to their corresponding RateLimitItem subclasses
67
    for programmatic rate limit item creation.
68
    
69
    Args:
70
        granularity_string: Granularity name like "second", "minute", "hour"
71
        
72
    Returns:
73
        RateLimitItem subclass matching the granularity
74
        
75
    Raises:
76
        ValueError: If no granularity matches the provided string
77
        
78
    Examples:
79
        granularity_from_string("second")  # Returns RateLimitItemPerSecond
80
        granularity_from_string("minute")  # Returns RateLimitItemPerMinute
81
        granularity_from_string("hour")    # Returns RateLimitItemPerHour
82
    """
83
```
84

85
### Window Statistics
86

87
Data structures for representing rate limiting window information and current quota status.
88

89
```python { .api }
90
from typing import NamedTuple
91

92
class WindowStats(NamedTuple):
93
    """
94
    Statistics for a rate limiting window.
95
    
96
    Provides information about the current state of a rate limit window,
97
    including when it will reset and how much quota remains available.
98
    """
99
    
100
    reset_time: float
101
    """
102
    Time when the current window will reset (seconds since Unix epoch).
103
    
104
    For fixed windows, this is when the current fixed window expires.
105
    For moving windows, this is when the oldest request in the window expires.
106
    For sliding windows, this is an approximation based on window algorithm.
107
    """
108
    
109
    remaining: int
110
    """
111
    Number of requests remaining in the current window.
112
    
113
    This represents the available quota before the rate limit would be
114
    exceeded. A value of 0 means the limit is fully consumed.
115
    """
116
```
117

118
### Dependency Management
119

120
Classes for handling optional dependencies and lazy loading of storage backend requirements.
121

122
```python { .api }
123
from typing import Dict, List, Union, Optional
124
from packaging.version import Version
125
from types import ModuleType
126

127
class Dependency:
128
    """Information about a single dependency"""
129
    name: str
130
    version_required: Optional[Version]
131
    version_found: Optional[Version] 
132
    module: ModuleType
133

134
class DependencyDict(dict):
135
    """
136
    Dictionary that validates dependencies when accessed.
137
    
138
    Extends dict to provide automatic dependency validation and helpful
139
    error messages when required dependencies are missing or have
140
    incompatible versions.
141
    """
142
    
143
    def __getitem__(self, key: str) -> Dependency:
144
        """
145
        Get dependency with validation.
146
        
147
        Args:
148
            key: Dependency name (module name)
149
            
150
        Returns:
151
            Dependency instance with validated module and version
152
            
153
        Raises:
154
            ConfigurationError: If dependency is missing or version incompatible
155
        """
156

157
class LazyDependency:
158
    """
159
    Utility for lazy loading of optional dependencies.
160
    
161
    Base class that provides dependency management for storage backends
162
    and other components that require optional dependencies. Dependencies
163
    are only imported when the storage is actually instantiated.
164
    """
165
    
166
    DEPENDENCIES: Union[Dict[str, Optional[Version]], List[str]] = []
167
    """
168
    Specification of required dependencies.
169
    
170
    Can be either:
171
    - List of dependency names: ["redis", "pymongo"]  
172
    - Dict mapping names to minimum versions: {"redis": Version("4.0.0")}
173
    """
174
    
175
    def __init__(self):
176
        """Initialize with empty dependency cache"""
177
        self._dependencies: DependencyDict = DependencyDict()
178
    
179
    @property
180
    def dependencies(self) -> DependencyDict:
181
        """
182
        Cached mapping of dependencies with lazy loading.
183
        
184
        Dependencies are imported and validated only when first accessed,
185
        allowing storage classes to be imported without requiring all
186
        their dependencies to be installed.
187
        
188
        Returns:
189
            DependencyDict with validated dependencies
190
        """
191

192
def get_dependency(module_path: str) -> tuple[ModuleType, Optional[Version]]:
193
    """
194
    Safely import a module at runtime.
195
    
196
    Attempts to import the specified module and determine its version,
197
    returning a placeholder if the import fails.
198
    
199
    Args:
200
        module_path: Full module path like "redis" or "pymongo.mongo_client"
201
        
202
    Returns:
203
        Tuple of (module, version) or (MissingModule, None) if import fails
204
    """
205
```
206

207
### Package Utilities
208

209
Functions for accessing package resources and handling internal utilities.
210

211
```python { .api }
212
def get_package_data(path: str) -> bytes:
213
    """
214
    Read data from package resources.
215
    
216
    Provides access to data files bundled with the limits package,
217
    such as Lua scripts for Redis operations or configuration templates.
218
    
219
    Args:
220
        path: Relative path within the limits package
221
        
222
    Returns:
223
        Raw bytes of the requested resource
224
        
225
    Examples:
226
        script_data = get_package_data("storage/redis_scripts/sliding_window.lua")
227
    """
228
```
229

230
### Internal Constants and Patterns
231

232
Regular expressions and constants used for parsing rate limit strings.
233

234
```python { .api }
235
import re
236

237
# Regular expression patterns for parsing rate limit strings
238
SEPARATORS: re.Pattern = re.compile(r"[,;|]{1}")
239
"""Pattern for separating multiple rate limits"""
240

241
SINGLE_EXPR: re.Pattern = re.compile(
242
    r"""
243
    \s*([0-9]+)
244
    \s*(/|\s*per\s*)
245
    \s*([0-9]+)
246
    *\s*(hour|minute|second|day|month|year)s?\s*""",
247
    re.IGNORECASE | re.VERBOSE
248
)
249
"""Pattern for matching a single rate limit expression"""
250

251
EXPR: re.Pattern = re.compile(
252
    rf"^{SINGLE_EXPR.pattern}(:?{SEPARATORS.pattern}{SINGLE_EXPR.pattern})*$",
253
    re.IGNORECASE | re.VERBOSE  
254
)
255
"""Pattern for matching one or more rate limit expressions"""
256
```
257

258
## Usage Examples
259

260
### Parsing Rate Limit Strings
261

262
```python
263
from limits.util import parse, parse_many, granularity_from_string
264

265
# Parse single rate limits
266
rate_limit_1 = parse("10/second")
267
print(type(rate_limit_1).__name__)  # RateLimitItemPerSecond
268
print(rate_limit_1.amount)          # 10
269

270
rate_limit_2 = parse("5 per minute")  
271
print(type(rate_limit_2).__name__)  # RateLimitItemPerMinute
272
print(rate_limit_2.amount)          # 5
273

274
# Parse multiple rate limits
275
multi_limits = parse_many("10/second; 100/minute; 1000/hour")
276
print(len(multi_limits))            # 3
277
print([rl.amount for rl in multi_limits])  # [10, 100, 1000]
278

279
# Different separators work
280
comma_limits = parse_many("5/second, 50/minute")
281
pipe_limits = parse_many("1 per second | 10 per minute")
282

283
# Get granularity classes programmatically
284
SecondClass = granularity_from_string("second")
285
MinuteClass = granularity_from_string("minute")
286

287
# Create instances using discovered classes
288
dynamic_limit = SecondClass(20)  # Same as RateLimitItemPerSecond(20)
289
```
290

291
### Working with Window Statistics
292

293
```python
294
from limits import RateLimitItemPerMinute
295
from limits.storage import MemoryStorage
296
from limits.strategies import FixedWindowRateLimiter
297
from limits.util import WindowStats
298
import time
299

300
# Setup rate limiting
301
rate_limit = RateLimitItemPerMinute(60)  # 60 requests per minute
302
storage = MemoryStorage()
303
limiter = FixedWindowRateLimiter(storage)
304

305
user_id = "user123"
306

307
# Make some requests
308
for i in range(10):
309
    limiter.hit(rate_limit, user_id)
310

311
# Get window statistics
312
stats: WindowStats = limiter.get_window_stats(rate_limit, user_id)
313

314
print(f"Remaining requests: {stats.remaining}")
315
print(f"Window resets at: {stats.reset_time}")
316
print(f"Time until reset: {stats.reset_time - time.time():.2f} seconds")
317

318
# Check if we can make more requests
319
if stats.remaining > 0:
320
    print(f"Can make {stats.remaining} more requests")
321
else:
322
    print("Rate limit exhausted")
323
    reset_in = stats.reset_time - time.time()
324
    print(f"Try again in {reset_in:.2f} seconds")
325
```
326

327
### Dependency Management Example
328

329
```python
330
from limits.util import LazyDependency, get_dependency
331
from limits.errors import ConfigurationError
332
from packaging.version import Version
333

334
class CustomStorage(LazyDependency):
335
    """Example storage with dependency requirements"""
336
    
337
    # Specify required dependencies with minimum versions
338
    DEPENDENCIES = {
339
        "redis": Version("4.0.0"),
340
        "requests": Version("2.25.0")
341
    }
342
    
343
    def __init__(self, uri: str):
344
        super().__init__()
345
        self.uri = uri
346
    
347
    def connect(self):
348
        """Connect using dependencies"""
349
        try:
350
            # Access dependencies - will validate and import
351
            redis_dep = self.dependencies["redis"]
352
            requests_dep = self.dependencies["requests"]
353
            
354
            # Use the validated modules
355
            redis_module = redis_dep.module
356
            requests_module = requests_dep.module
357
            
358
            print(f"Using Redis version: {redis_dep.version_found}")
359
            print(f"Using Requests version: {requests_dep.version_found}")
360
            
361
        except ConfigurationError as e:
362
            print(f"Dependency error: {e}")
363

364
# Test manual dependency checking
365
redis_module, redis_version = get_dependency("redis")
366
if redis_module.__name__ != "Missing":
367
    print(f"Redis is available, version: {redis_version}")
368
else:
369
    print("Redis is not installed")
370
```
371

372
### Practical Parsing Scenarios
373

374
```python
375
from limits.util import parse_many
376
from limits.storage import MemoryStorage  
377
from limits.strategies import FixedWindowRateLimiter
378

379
def setup_api_rate_limiting():
380
    """Setup multi-tier rate limiting from configuration"""
381
    
382
    # Configuration from environment or config file
383
    rate_limit_config = "10/second; 100/minute; 1000/hour; 5000/day"
384
    
385
    # Parse all rate limits
386
    rate_limits = parse_many(rate_limit_config)
387
    
388
    # Setup storage and limiters
389
    storage = MemoryStorage()
390
    limiters = [FixedWindowRateLimiter(storage) for _ in rate_limits]
391
    
392
    return list(zip(rate_limits, limiters))
393

394
def check_api_limits(user_id: str, rate_limit_pairs):
395
    """Check all rate limits for a user"""
396
    
397
    for rate_limit, limiter in rate_limit_pairs:
398
        if not limiter.test(rate_limit, user_id):
399
            # Find which limit was exceeded
400
            granularity = rate_limit.GRANULARITY.name
401
            amount = rate_limit.amount
402
            print(f"Rate limit exceeded: {amount} per {granularity}")
403
            return False
404
    
405
    # All limits passed, consume from all
406
    for rate_limit, limiter in rate_limit_pairs:
407
        limiter.hit(rate_limit, user_id)
408
    
409
    return True
410

411
# Usage
412
rate_limiting_setup = setup_api_rate_limiting()
413
user_allowed = check_api_limits("user123", rate_limiting_setup)
414
print(f"User request allowed: {user_allowed}")
415
```
416

417
### Error Handling with Parsing
418

419
```python
420
from limits.util import parse, parse_many
421
from limits.errors import ConfigurationError
422

423
def safe_parse_limits(limit_strings: list[str]):
424
    """Safely parse rate limit strings with error handling"""
425
    
426
    valid_limits = []
427
    errors = []
428
    
429
    for limit_string in limit_strings:
430
        try:
431
            if ";" in limit_string or "," in limit_string or "|" in limit_string:
432
                # Multiple limits
433
                limits = parse_many(limit_string)
434
                valid_limits.extend(limits)
435
            else:
436
                # Single limit
437
                limit = parse(limit_string)
438
                valid_limits.append(limit)
439
                
440
        except ValueError as e:
441
            errors.append(f"Invalid rate limit '{limit_string}': {e}")
442
    
443
    return valid_limits, errors
444

445
# Test with mixed valid and invalid strings
446
test_limits = [
447
    "10/second",           # Valid
448
    "100 per minute",      # Valid
449
    "invalid/format",      # Invalid
450
    "5/second; 50/minute", # Valid multiple
451
    "bad; worse",          # Invalid multiple
452
]
453

454
valid, errors = safe_parse_limits(test_limits)
455
print(f"Valid limits: {len(valid)}")
456
print(f"Errors: {errors}")
457
```