Utilities

def parse(limit_string: str) -> RateLimitItem:
    """
    Parse a single rate limit string into a RateLimitItem.
    
    Converts human-readable rate limit notation into the appropriate
    RateLimitItem subclass based on the granularity specified.
    
    Args:
        limit_string: Rate limit string like "10/second", "5 per minute", 
                     "100/hour", "1000 per day"
    
    Returns:
        RateLimitItem instance matching the specified granularity
        
    Raises:
        ValueError: If the string notation is invalid or unparseable
        
    Examples:
        parse("10/second")     # Returns RateLimitItemPerSecond(10)
        parse("5 per minute")  # Returns RateLimitItemPerMinute(5)
        parse("100/hour")      # Returns RateLimitItemPerHour(100)
        parse("1000 per day")  # Returns RateLimitItemPerDay(1000)
    """

def parse_many(limit_string: str) -> list[RateLimitItem]:
    """
    Parse multiple rate limit strings separated by delimiters.
    
    Supports comma, semicolon, or pipe-separated rate limit specifications,
    allowing complex multi-tier rate limiting configurations in a single string.
    
    Args:
        limit_string: Multiple rate limits like "10/second; 100/minute; 1000/hour"
        
    Returns:
        List of RateLimitItem instances for each parsed rate limit
        
    Raises:
        ValueError: If any part of the string notation is invalid
        
    Examples:
        parse_many("10/second; 100/minute")
        # Returns [RateLimitItemPerSecond(10), RateLimitItemPerMinute(100)]
        
        parse_many("5/second, 50/minute, 500/hour")  
        # Returns [RateLimitItemPerSecond(5), RateLimitItemPerMinute(50), RateLimitItemPerHour(500)]
        
        parse_many("1 per second | 10 per minute")
        # Returns [RateLimitItemPerSecond(1), RateLimitItemPerMinute(10)]
    """

def granularity_from_string(granularity_string: str) -> type[RateLimitItem]:
    """
    Get RateLimitItem class for a granularity string.
    
    Maps granularity names to their corresponding RateLimitItem subclasses
    for programmatic rate limit item creation.
    
    Args:
        granularity_string: Granularity name like "second", "minute", "hour"
        
    Returns:
        RateLimitItem subclass matching the granularity
        
    Raises:
        ValueError: If no granularity matches the provided string
        
    Examples:
        granularity_from_string("second")  # Returns RateLimitItemPerSecond
        granularity_from_string("minute")  # Returns RateLimitItemPerMinute
        granularity_from_string("hour")    # Returns RateLimitItemPerHour
    """

from typing import NamedTuple

class WindowStats(NamedTuple):
    """
    Statistics for a rate limiting window.
    
    Provides information about the current state of a rate limit window,
    including when it will reset and how much quota remains available.
    """
    
    reset_time: float
    """
    Time when the current window will reset (seconds since Unix epoch).
    
    For fixed windows, this is when the current fixed window expires.
    For moving windows, this is when the oldest request in the window expires.
    For sliding windows, this is an approximation based on window algorithm.
    """
    
    remaining: int
    """
    Number of requests remaining in the current window.
    
    This represents the available quota before the rate limit would be
    exceeded. A value of 0 means the limit is fully consumed.
    """

from typing import Dict, List, Union, Optional
from packaging.version import Version
from types import ModuleType

class Dependency:
    """Information about a single dependency"""
    name: str
    version_required: Optional[Version]
    version_found: Optional[Version] 
    module: ModuleType

class DependencyDict(dict):
    """
    Dictionary that validates dependencies when accessed.
    
    Extends dict to provide automatic dependency validation and helpful
    error messages when required dependencies are missing or have
    incompatible versions.
    """
    
    def __getitem__(self, key: str) -> Dependency:
        """
        Get dependency with validation.
        
        Args:
            key: Dependency name (module name)
            
        Returns:
            Dependency instance with validated module and version
            
        Raises:
            ConfigurationError: If dependency is missing or version incompatible
        """

class LazyDependency:
    """
    Utility for lazy loading of optional dependencies.
    
    Base class that provides dependency management for storage backends
    and other components that require optional dependencies. Dependencies
    are only imported when the storage is actually instantiated.
    """
    
    DEPENDENCIES: Union[Dict[str, Optional[Version]], List[str]] = []
    """
    Specification of required dependencies.
    
    Can be either:
    - List of dependency names: ["redis", "pymongo"]  
    - Dict mapping names to minimum versions: {"redis": Version("4.0.0")}
    """
    
    def __init__(self):
        """Initialize with empty dependency cache"""
        self._dependencies: DependencyDict = DependencyDict()
    
    @property
    def dependencies(self) -> DependencyDict:
        """
        Cached mapping of dependencies with lazy loading.
        
        Dependencies are imported and validated only when first accessed,
        allowing storage classes to be imported without requiring all
        their dependencies to be installed.
        
        Returns:
            DependencyDict with validated dependencies
        """

def get_dependency(module_path: str) -> tuple[ModuleType, Optional[Version]]:
    """
    Safely import a module at runtime.
    
    Attempts to import the specified module and determine its version,
    returning a placeholder if the import fails.
    
    Args:
        module_path: Full module path like "redis" or "pymongo.mongo_client"
        
    Returns:
        Tuple of (module, version) or (MissingModule, None) if import fails
    """

def get_package_data(path: str) -> bytes:
    """
    Read data from package resources.
    
    Provides access to data files bundled with the limits package,
    such as Lua scripts for Redis operations or configuration templates.
    
    Args:
        path: Relative path within the limits package
        
    Returns:
        Raw bytes of the requested resource
        
    Examples:
        script_data = get_package_data("storage/redis_scripts/sliding_window.lua")
    """

import re

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

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

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

from limits.util import parse, parse_many, granularity_from_string

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

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

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

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

# Get granularity classes programmatically
SecondClass = granularity_from_string("second")
MinuteClass = granularity_from_string("minute")

# Create instances using discovered classes
dynamic_limit = SecondClass(20)  # Same as RateLimitItemPerSecond(20)

from limits import RateLimitItemPerMinute
from limits.storage import MemoryStorage
from limits.strategies import FixedWindowRateLimiter
from limits.util import WindowStats
import time

# Setup rate limiting
rate_limit = RateLimitItemPerMinute(60)  # 60 requests per minute
storage = MemoryStorage()
limiter = FixedWindowRateLimiter(storage)

user_id = "user123"

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

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

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

# Check if we can make more requests
if stats.remaining > 0:
    print(f"Can make {stats.remaining} more requests")
else:
    print("Rate limit exhausted")
    reset_in = stats.reset_time - time.time()
    print(f"Try again in {reset_in:.2f} seconds")

from limits.util import LazyDependency, get_dependency
from limits.errors import ConfigurationError
from packaging.version import Version

class CustomStorage(LazyDependency):
    """Example storage with dependency requirements"""
    
    # Specify required dependencies with minimum versions
    DEPENDENCIES = {
        "redis": Version("4.0.0"),
        "requests": Version("2.25.0")
    }
    
    def __init__(self, uri: str):
        super().__init__()
        self.uri = uri
    
    def connect(self):
        """Connect using dependencies"""
        try:
            # Access dependencies - will validate and import
            redis_dep = self.dependencies["redis"]
            requests_dep = self.dependencies["requests"]
            
            # Use the validated modules
            redis_module = redis_dep.module
            requests_module = requests_dep.module
            
            print(f"Using Redis version: {redis_dep.version_found}")
            print(f"Using Requests version: {requests_dep.version_found}")
            
        except ConfigurationError as e:
            print(f"Dependency error: {e}")

# Test manual dependency checking
redis_module, redis_version = get_dependency("redis")
if redis_module.__name__ != "Missing":
    print(f"Redis is available, version: {redis_version}")
else:
    print("Redis is not installed")

from limits.util import parse_many
from limits.storage import MemoryStorage  
from limits.strategies import FixedWindowRateLimiter

def setup_api_rate_limiting():
    """Setup multi-tier rate limiting from configuration"""
    
    # Configuration from environment or config file
    rate_limit_config = "10/second; 100/minute; 1000/hour; 5000/day"
    
    # Parse all rate limits
    rate_limits = parse_many(rate_limit_config)
    
    # Setup storage and limiters
    storage = MemoryStorage()
    limiters = [FixedWindowRateLimiter(storage) for _ in rate_limits]
    
    return list(zip(rate_limits, limiters))

def check_api_limits(user_id: str, rate_limit_pairs):
    """Check all rate limits for a user"""
    
    for rate_limit, limiter in rate_limit_pairs:
        if not limiter.test(rate_limit, user_id):
            # Find which limit was exceeded
            granularity = rate_limit.GRANULARITY.name
            amount = rate_limit.amount
            print(f"Rate limit exceeded: {amount} per {granularity}")
            return False
    
    # All limits passed, consume from all
    for rate_limit, limiter in rate_limit_pairs:
        limiter.hit(rate_limit, user_id)
    
    return True

# Usage
rate_limiting_setup = setup_api_rate_limiting()
user_allowed = check_api_limits("user123", rate_limiting_setup)
print(f"User request allowed: {user_allowed}")

from limits.util import parse, parse_many
from limits.errors import ConfigurationError

def safe_parse_limits(limit_strings: list[str]):
    """Safely parse rate limit strings with error handling"""
    
    valid_limits = []
    errors = []
    
    for limit_string in limit_strings:
        try:
            if ";" in limit_string or "," in limit_string or "|" in limit_string:
                # Multiple limits
                limits = parse_many(limit_string)
                valid_limits.extend(limits)
            else:
                # Single limit
                limit = parse(limit_string)
                valid_limits.append(limit)
                
        except ValueError as e:
            errors.append(f"Invalid rate limit '{limit_string}': {e}")
    
    return valid_limits, errors

# Test with mixed valid and invalid strings
test_limits = [
    "10/second",           # Valid
    "100 per minute",      # Valid
    "invalid/format",      # Invalid
    "5/second; 50/minute", # Valid multiple
    "bad; worse",          # Invalid multiple
]

valid, errors = safe_parse_limits(test_limits)
print(f"Valid limits: {len(valid)}")
print(f"Errors: {errors}")