Skip to content

API Reference

API Reference

Quick links:

pycurb.core.models

pycurb.core.models.LimitRule

Bases: BaseModel

Configuration for a rate limit rule.

Users create an instance of this class to define how requests are limited.

Attributes:

Name Type Description
name str

Short unique identifier for the rule.
algorithm Literal['sliding_window', 'fixed_window', 'token_bucket', 'leaky_bucket', 'gcra']

Rate limit algorithm to use.
limit Optional[int]

Maximum requests allowed (window-based) or capacity (token bucket/gcra).
window Optional[int]

Time window in seconds (for window-based algorithms) or base to calculate refill rate.
capacity Optional[int]

Maximum capacity for token bucket or gcra (burst limit).
refill_rate Optional[float]

Tokens added per second for token bucket or requests allowed per second for GCRA.
leak_rate Optional[float]

Requests processed per second.
metadata Dict[str, Any]

Arbitrary user data for extra conditional logic.

Methods:

validate_algorithm_parameters 

validate_algorithm_parameters()

Ensure algorithm-specific parameters are valid.

Source code in src/pycurb/core/models.py 
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
@model_validator(mode="after")
def validate_algorithm_parameters(self) -> "LimitRule":
    """
    Ensure algorithm-specific parameters are valid.
    """
    if self.algorithm in ("sliding_window", "fixed_window"):
        if self.limit is None or self.window is None:
            raise ValueError(
                f"'limit' and 'window' are required for {self.algorithm} algorithm."
            )

    elif self.algorithm == "token_bucket":
        has_cap = self.capacity is not None
        has_refill = self.refill_rate is not None
        has_limit = self.limit is not None
        has_window = self.window is not None

        # Must have capacity (or limit) for bucket size
        if not (has_cap or has_limit):
            raise ValueError(
                "'capacity' or 'limit' is required for token_bucket algorithm."
            )

        # Must have refill_rate (or window) to determine how tokens are added
        if not (has_refill or has_window):
            raise ValueError(
                "'refill_rate' or 'window' is required for token_bucket algorithm."
            )

        cap = self.capacity if has_cap else self.limit
        if cap is None:
            raise ValueError(
                "'capacity' or 'limit' is required for token_bucket algorithm."
            )

        if self.refill_rate is not None:
            rate = self.refill_rate
        else:
            if self.window is None:
                raise ValueError(
                    "'window' is required for token_bucket algorithm when 'refill_rate' is not provided."
                )
            rate = cap / self.window

        if rate <= 0:
            raise ValueError(f"'refill_rate' must be positive, got {rate}")

    elif self.algorithm == "leaky_bucket":
        has_cap = self.capacity is not None
        has_limit = self.limit is not None
        has_window = self.window is not None

        if not (has_cap or has_limit):
            raise ValueError(
                "'capacity' or 'limit' is required for leaky_bucket algorithm."
            )

        cap = self.capacity if self.capacity is not None else self.limit

        if cap is None:
            raise ValueError(
                "'capacity' or 'limit' is required for leaky_bucket algorithm."
            )

        if self.leak_rate is not None:
            rate = self.leak_rate
        else:
            if self.window is None:
                raise ValueError(
                    "'window' or 'leak_rate' is required for leaky_bucket algorithm."
                )
            rate = cap / self.window

        if rate <= 0:
            raise ValueError(f"'leak_rate' must be positive, got {rate}")

    elif self.algorithm == "gcra":
        has_cap = self.capacity is not None
        has_limit = self.limit is not None

        # Must have capacity (or limit) for maximum burst capacity
        if not (has_cap or has_limit):
            raise ValueError(
                "'capacity' or 'limit' is required for Gcra algorithm's burst capacity."
            )

        cap = self.capacity if has_cap else self.limit
        if cap is None:
            raise ValueError(
                "'capacity' or 'limit' is required for Gcra algorithm's burst capacity."
            )

        # Must have refill_rate for requests allowed per second
        if self.refill_rate is not None:
            rate = self.refill_rate
        else:
            if self.window is None:
                raise ValueError(
                    "'window' is required for Gcra algorithm's requests per second if 'refill_rate' is not provided."
                )
            rate = cap / self.window

        if rate <= 0:
            raise ValueError(f"'refill_rate' must be positive, got {rate}")

    return self

pycurb.core.models.RateLimitResult

Bases: BaseModel

Outcome of a rate limit check.

Attributes:

Name Type Description
allowed bool

Request permitted or not.
remaining int

Remaining requests/tokens in current window/bucket.
reset_at float

Unix timestamp (seconds) when the limit resets.
limit int

The configured limit (for X-RateLimit-Limit header).
retry_after Optional[int]

Seconds to wait before retrying (only when allowed is False).
rule_name Optional[str]

Name of the rule that was applied.

pycurb.core.models.RateLimitHeaders

Bases: BaseModel

Helper to build standard HTTP ratelimit headers.

Attributes:

Name Type Description
limit int

Value for X-RateLimit-Limit.
remaining int

Value for X-RateLimit-Remaining.
reset int

Value for X-RateLimit-Reset (unix timestamp).
retry_after Optional[int]

Optional Retry-After value in seconds.

Methods:

from_result classmethod 

from_result(result, now=None)

Construct headers from a rate limit result If retry_after is not set, compute as max(0, reset_at - now).

Source code in src/pycurb/core/models.py 
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
@classmethod
def from_result(
    cls, result: RateLimitResult, now: Optional[float] = None
) -> "RateLimitHeaders":
    """
    Construct headers from a rate limit result
    If retry_after is not set, compute as max(0, reset_at - now).
    """
    if now is None:
        now = time.time()
    retry_after = result.retry_after
    if retry_after is None and not result.allowed:
        retry_after = max(0, math.ceil(result.reset_at - now))

    return cls(
        limit=result.limit,
        remaining=result.remaining,
        reset=int(result.reset_at),
        retry_after=retry_after,
    )

to_dict 

to_dict()

Convert to dictionary of header names and values

Source code in src/pycurb/core/models.py 
228
229
230
231
232
233
234
235
236
237
def to_dict(self) -> Dict[str, str]:
    """Convert to dictionary of header names and values"""
    headers = {
        "X-RateLimit-Limit": str(self.limit),
        "X-RateLimit-Remaining": str(self.remaining),
        "X-RateLimit-Reset": str(self.reset),
    }
    if self.retry_after is not None:
        headers["Retry-After"] = str(self.retry_after)
    return headers

pycurb.core.models.RateLimitExceeded 

RateLimitExceeded(result)

Bases: Exception

Exception raised when rate limit has been exceeded.

Attributes:

Name Type Description
result

The RateLimitResult instance returned by the limiter containing details.
Source code in src/pycurb/core/models.py 
268
269
270
271
def __init__(self, result: RateLimitResult):
    self.result = result
    retry_after = max(0, math.ceil(result.reset_at - time.time()))
    super().__init__(f"Rate limit exceeded. Retry after {retry_after} seconds.")

pycurb.core.storage

pycurb.core.storage.base.Storage

Bases: ABC

Abstract storage backend for rate limiter counters.

Methods:

close abstractmethod 

close()

Release any resources (connections, etc.).

Source code in src/pycurb/core/storage/base.py 
120
121
122
123
@abstractmethod
def close(self) -> None:
    """Release any resources (connections, etc.)."""
    pass

fixed_window abstractmethod 

fixed_window(key, window, limit, now)

Record a request using fixed window algorithm.

The window is aligned to calendar boundaries: e.g., window=60 means windows start at timestamps 0, 60, 120, ... (floor(now / window) * window).

Parameters:

Name Type Description Default
key str

Unique identifier.

 required
window int

Window size in seconds.

 required
limit int

Maximum requests per window.

 required
now float

Current Unix timestamp (seconds).

 required

Returns:

Type Description
bool

Tuple (allowed, remaining, reset_at)
int
  • allowed: bool - True if request within limit.
float
  • remaining: int - Remaining requests allowed in current window.
Tuple[bool, int, float]
  • reset_at: start of next window (aligned).
Source code in src/pycurb/core/storage/base.py 
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
@abstractmethod
def fixed_window(
    self, key: str, window: int, limit: int, now: float
) -> Tuple[bool, int, float]:
    """
    Record a request using fixed window algorithm.

    The window is aligned to calendar boundaries: e.g., window=60 means
    windows start at timestamps 0, 60, 120, ... (floor(now / window) * window).

    Args:
        key: Unique identifier.
        window: Window size in seconds.
        limit: Maximum requests per window.
        now: Current Unix timestamp (seconds).

    Returns:
        Tuple (allowed, remaining, reset_at)
        - allowed: bool - True if request within limit.
        - remaining: int - Remaining requests allowed in current window.
        - reset_at: start of next window (aligned).
    """
    pass

gcra abstractmethod 

gcra(key, capacity, rate, now)

Process a request through Generic Cell Rate Algorithm. It enforces a steady request rate with a defined burst limit.

Parameters:

Name Type Description Default
key str

Unique identifier.

 required
capacity int

Maximum burst size (number of requests that can be sent immediately).

 required
rate float

Requests processed per second.

 required
now float

Current Unix timestamp (seconds).

 required

Returns:

Type Description
bool

(allowed, remaining_capacity, reset_at)
int
  • allowed: True if request is within the rate and burst limit.
float
  • remaining: Remaining burst capacity (How many additional requests can be sent without waiting).
Tuple[bool, int, float]
  • reset_at: Timestamp when the next request would be allowed.
Source code in src/pycurb/core/storage/base.py 
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
@abstractmethod
def gcra(
    self, key: str, capacity: int, rate: float, now: float
) -> Tuple[bool, int, float]:
    """
    Process a request through Generic Cell Rate Algorithm.
    It enforces a steady request rate with a defined burst limit.

    Args:
        key: Unique identifier.
        capacity: Maximum burst size (number of requests that can be sent immediately).
        rate: Requests processed per second.
        now: Current Unix timestamp (seconds).

    Returns:
        (allowed, remaining_capacity, reset_at)
        - allowed: True if request is within the rate and burst limit.
        - remaining: Remaining burst capacity (How many additional requests can be sent without waiting).
        - reset_at: Timestamp when the next request would be allowed.
    """
    pass

leaky_bucket abstractmethod 

leaky_bucket(key, capacity, leak_rate, now)

Process a request through a leaky bucket.

The bucket has a capacity (queue size). Requests leak out at leak_rate per second. If the bucket is full, request is denied.

Parameters:

Name Type Description Default
key str

Unique identifier.

 required
capacity int

Maximum queue size.

 required
leak_rate float

Requests processed per second (leak rate).

 required
now float

Current Unix timestamp (seconds).

 required

Returns:

Type Description
bool

(allowed, remaining_capacity, reset_at)
int
  • allowed: True if request can be enqueued.
float
  • remaining_capacity: Slots left in the queue.
Tuple[bool, int, float]
  • reset_at: Timestamp when the next request can be processed (if full).
Source code in src/pycurb/core/storage/base.py 
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
@abstractmethod
def leaky_bucket(
    self, key: str, capacity: int, leak_rate: float, now: float
) -> Tuple[bool, int, float]:
    """
    Process a request through a leaky bucket.

    The bucket has a capacity (queue size). Requests leak out at leak_rate per second.
    If the bucket is full, request is denied.

    Args:
        key: Unique identifier.
        capacity: Maximum queue size.
        leak_rate: Requests processed per second (leak rate).
        now: Current Unix timestamp (seconds).

    Returns:
        (allowed, remaining_capacity, reset_at)
        - allowed: True if request can be enqueued.
        - remaining_capacity: Slots left in the queue.
        - reset_at: Timestamp when the next request can be processed (if full).
    """
    pass

sliding_window abstractmethod 

sliding_window(key, window, limit, now)

Record a request using sliding window algorithm.

Parameters:

Name Type Description Default
key str

Unique identifier.

 required
window int

Time window in seconds.

 required
limit int

Maximum allowed requests per window.

 required
now float

Current Unix timestamp (seconds).

 required

Returns:

Type Description
bool

Tuple (allowed, remaining, reset_at)
int
  • allowed: bool - True if request within limit.
float
  • remaining: int - Remaining requests allowed in current window.
Tuple[bool, int, float]
  • reset_at: float - Unix timestamp when the window expires (oldest request + window).
Source code in src/pycurb/core/storage/base.py 
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
@abstractmethod
def sliding_window(
    self, key: str, window: int, limit: int, now: float
) -> Tuple[bool, int, float]:
    """
    Record a request using sliding window algorithm.

    Args:
        key: Unique identifier.
        window: Time window in seconds.
        limit: Maximum allowed requests per window.
        now: Current Unix timestamp (seconds).

    Returns:
        Tuple (allowed, remaining, reset_at)
        - allowed: bool - True if request within limit.
        - remaining: int - Remaining requests allowed in current window.
        - reset_at: float - Unix timestamp when the window expires (oldest request + window).
    """
    pass

token_bucket abstractmethod 

token_bucket(key, capacity, refill_rate, now)

Consume one token from a token bucket.

Parameters:

Name Type Description Default
key str

Unique identifier.

 required
capacity int

Maximum tokens (burst limit).

 required
refill_rate float

Tokens added per second.

 required
now float

Current Unix timestamp (seconds).

 required

Returns:

Type Description
bool

(allowed, remaining_tokens, reset_at)
int
  • allowed: True if a token was available.
float
  • remaining_tokens: Tokens left after consumption (or current tokens if not allowed).
Tuple[bool, int, float]
  • reset_at: Timestamp when next token will be added (if not allowed) or when bucket would be full (optional). Usually used for Retry-After.
Source code in src/pycurb/core/storage/base.py 
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
@abstractmethod
def token_bucket(
    self, key: str, capacity: int, refill_rate: float, now: float
) -> Tuple[bool, int, float]:
    """
    Consume one token from a token bucket.

    Args:
        key: Unique identifier.
        capacity: Maximum tokens (burst limit).
        refill_rate: Tokens added per second.
        now: Current Unix timestamp (seconds).

    Returns:
        (allowed, remaining_tokens, reset_at)
        - allowed: True if a token was available.
        - remaining_tokens: Tokens left after consumption (or current tokens if not allowed).
        - reset_at: Timestamp when next token will be added (if not allowed) or when bucket would be full (optional). Usually used for Retry-After.
    """
    pass

pycurb.core.storage.base_async.AsyncStorage

Bases: ABC

Abstract storage backend for rate limiter counters. Asynchronous version (WSGI compatible)

Methods:

close abstractmethod async 

close()

Release any resources (connections, etc.).

Source code in src/pycurb/core/storage/base_async.py 
120
121
122
123
@abstractmethod
async def close(self) -> None:
    """Release any resources (connections, etc.)."""
    pass

fixed_window abstractmethod async 

fixed_window(key, window, limit, now)

Record a request using fixed window algorithm.

The window is aligned to calendar boundaries: e.g., window=60 means windows start at timestamps 0, 60, 120, ... (floor(now / window) * window).

Parameters:

Name Type Description Default
key str

Unique identifier.

 required
window int

Window size in seconds.

 required
limit int

Maximum requests per window.

 required
now float

Current Unix timestamp (seconds).

 required

Returns:

Type Description
bool

Tuple (allowed, remaining, reset_at)
int
  • allowed: bool - True if request within limit.
float
  • remaining: int - Remaining requests allowed in current window.
Tuple[bool, int, float]
  • reset_at: start of next window (aligned).
Source code in src/pycurb/core/storage/base_async.py 
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
@abstractmethod
async def fixed_window(
    self, key: str, window: int, limit: int, now: float
) -> Tuple[bool, int, float]:
    """
    Record a request using fixed window algorithm.

    The window is aligned to calendar boundaries: e.g., window=60 means
    windows start at timestamps 0, 60, 120, ... (floor(now / window) * window).

    Args:
        key: Unique identifier.
        window: Window size in seconds.
        limit: Maximum requests per window.
        now: Current Unix timestamp (seconds).

    Returns:
        Tuple (allowed, remaining, reset_at)
        - allowed: bool - True if request within limit.
        - remaining: int - Remaining requests allowed in current window.
        - reset_at: start of next window (aligned).
    """
    pass

gcra abstractmethod async 

gcra(key, capacity, rate, now)

Process a request through Generic Cell Rate Algorithm. It enforces a steady request rate with a defined burst limit.

Parameters:

Name Type Description Default
key str

Unique identifier.

 required
capacity int

Maximum burst size (number of requests that can be sent immediately).

 required
rate float

Requests processed per second.

 required
now float

Current Unix timestamp (seconds).

 required

Returns:

Type Description
bool

(allowed, remaining_capacity, reset_at)
int
  • allowed: True if request is within the rate and burst limit.
float
  • remaining: Remaining burst capacity (How many additional requests can be sent without waiting).
Tuple[bool, int, float]
  • reset_at: Timestamp when the next request would be allowed.
Source code in src/pycurb/core/storage/base_async.py 
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
@abstractmethod
async def gcra(
    self, key: str, capacity: int, rate: float, now: float
) -> Tuple[bool, int, float]:
    """
    Process a request through Generic Cell Rate Algorithm.
    It enforces a steady request rate with a defined burst limit.

    Args:
        key: Unique identifier.
        capacity: Maximum burst size (number of requests that can be sent immediately).
        rate: Requests processed per second.
        now: Current Unix timestamp (seconds).

    Returns:
        (allowed, remaining_capacity, reset_at)
        - allowed: True if request is within the rate and burst limit.
        - remaining: Remaining burst capacity (How many additional requests can be sent without waiting).
        - reset_at: Timestamp when the next request would be allowed.
    """
    pass

leaky_bucket abstractmethod async 

leaky_bucket(key, capacity, leak_rate, now)

Process a request through a leaky bucket.

The bucket has a capacity (queue size). Requests leak out at leak_rate per second. If the bucket is full, request is denied.

Parameters:

Name Type Description Default
key str

Unique identifier.

 required
capacity int

Maximum queue size.

 required
leak_rate float

Requests processed per second (leak rate).

 required
now float

Current Unix timestamp (seconds).

 required

Returns:

Type Description
bool

(allowed, remaining_capacity, reset_at)
int
  • allowed: True if request can be enqueued.
float
  • remaining_capacity: Slots left in the queue.
Tuple[bool, int, float]
  • reset_at: Timestamp when the next request can be processed (if full).
Source code in src/pycurb/core/storage/base_async.py 
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
@abstractmethod
async def leaky_bucket(
    self, key: str, capacity: int, leak_rate: float, now: float
) -> Tuple[bool, int, float]:
    """
    Process a request through a leaky bucket.

    The bucket has a capacity (queue size). Requests leak out at leak_rate per second.
    If the bucket is full, request is denied.

    Args:
        key: Unique identifier.
        capacity: Maximum queue size.
        leak_rate: Requests processed per second (leak rate).
        now: Current Unix timestamp (seconds).

    Returns:
        (allowed, remaining_capacity, reset_at)
        - allowed: True if request can be enqueued.
        - remaining_capacity: Slots left in the queue.
        - reset_at: Timestamp when the next request can be processed (if full).
    """
    pass

sliding_window abstractmethod async 

sliding_window(key, window, limit, now)

Record a request using sliding window algorithm.

Parameters:

Name Type Description Default
key str

Unique identifier.

 required
window int

Time window in seconds.

 required
limit int

Maximum allowed requests per window.

 required
now float

Current Unix timestamp (seconds).

 required

Returns:

Type Description
bool

Tuple (allowed, remaining, reset_at)
int
  • allowed: bool - True if request within limit.
float
  • remaining: int - Remaining requests allowed in current window.
Tuple[bool, int, float]
  • reset_at: float - Unix timestamp when the window expires (oldest request + window).
Source code in src/pycurb/core/storage/base_async.py 
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
@abstractmethod
async def sliding_window(
    self, key: str, window: int, limit: int, now: float
) -> Tuple[bool, int, float]:
    """
    Record a request using sliding window algorithm.

    Args:
        key: Unique identifier.
        window: Time window in seconds.
        limit: Maximum allowed requests per window.
        now: Current Unix timestamp (seconds).

    Returns:
        Tuple (allowed, remaining, reset_at)
        - allowed: bool - True if request within limit.
        - remaining: int - Remaining requests allowed in current window.
        - reset_at: float - Unix timestamp when the window expires (oldest request + window).
    """
    pass

token_bucket abstractmethod async 

token_bucket(key, capacity, refill_rate, now)

Consume one token from a token bucket.

Parameters:

Name Type Description Default
key str

Unique identifier.

 required
capacity int

Maximum tokens (burst limit).

 required
refill_rate float

Tokens added per second.

 required
now float

Current Unix timestamp (seconds).

 required

Returns:

Type Description
bool

(allowed, remaining_tokens, reset_at)
int
  • allowed: True if a token was available.
float
  • remaining_tokens: Tokens left after consumption (or current tokens if not allowed).
Tuple[bool, int, float]
  • reset_at: Timestamp when next token will be added (if not allowed) or when bucket would be full (optional). Usually used for Retry-After.
Source code in src/pycurb/core/storage/base_async.py 
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
@abstractmethod
async def token_bucket(
    self, key: str, capacity: int, refill_rate: float, now: float
) -> Tuple[bool, int, float]:
    """
    Consume one token from a token bucket.

    Args:
        key: Unique identifier.
        capacity: Maximum tokens (burst limit).
        refill_rate: Tokens added per second.
        now: Current Unix timestamp (seconds).

    Returns:
        (allowed, remaining_tokens, reset_at)
        - allowed: True if a token was available.
        - remaining_tokens: Tokens left after consumption (or current tokens if not allowed).
        - reset_at: Timestamp when next token will be added (if not allowed) or when bucket would be full (optional). Usually used for Retry-After.
    """
    pass

pycurb.core.storage.memory.MemoryStorage 

MemoryStorage()

Bases: Storage

In-memory storage for rate limiting.

Source code in src/pycurb/core/storage/memory.py 
11
12
13
14
15
16
17
18
def __init__(self) -> None:
    self._sliding: Dict[str, Deque[float]] = {}
    self._fixed: Dict[str, Tuple[int, float]] = {}
    self._token: Dict[str, Tuple[float, float]] = {}
    self._leaky: Dict[str, Tuple[int, float]] = {}
    self._gcra: Dict[str, float] = {}
    self._key_locks: Dict[str, threading.Lock] = {}
    self._global_lock = threading.Lock()

Methods:

get_lock 

get_lock(key)

Get or create a lock for the given key.

Source code in src/pycurb/core/storage/memory.py 
20
21
22
23
24
25
26
27
def get_lock(self, key: str) -> threading.Lock:
    """Get or create a lock for the given key."""
    if key in self._key_locks:
        return self._key_locks[key]
    with self._global_lock:
        if key not in self._key_locks:
            self._key_locks[key] = threading.Lock()
    return self._key_locks[key]

pycurb.core.storage.memory_async.AsyncMemoryStorage 

AsyncMemoryStorage()

Bases: AsyncStorage

Async in-memory storage for rate limiting.

Source code in src/pycurb/core/storage/memory_async.py 
11
12
13
14
15
16
17
18
def __init__(self) -> None:
    self._sliding: Dict[str, Deque[float]] = {}
    self._fixed: Dict[str, Tuple[int, float]] = {}
    self._token: Dict[str, Tuple[float, float]] = {}
    self._leaky: Dict[str, Tuple[int, float]] = {}
    self._gcra: Dict[str, float] = {}
    self._key_locks: Dict[str, asyncio.Lock] = {}
    self._global_lock = asyncio.Lock()

Methods:

get_lock async 

get_lock(key)

Get or create a lock for the given key.

Source code in src/pycurb/core/storage/memory_async.py 
20
21
22
23
24
25
26
27
async def get_lock(self, key: str) -> asyncio.Lock:
    """Get or create a lock for the given key."""
    if key in self._key_locks:
        return self._key_locks[key]
    async with self._global_lock:
        if key not in self._key_locks:
            self._key_locks[key] = asyncio.Lock()
    return self._key_locks[key]

pycurb.core.storage.redis.RedisStorage 

RedisStorage(
    redis_client,
    key_prefix="ratelimit:",
    use_redis_time=False,
    fallback_storage=None,
    fail_open=False,
)

Bases: Storage

Redis storage for rate limiting.

Source code in src/pycurb/core/storage/redis.py 
61
62
63
64
65
66
67
68
69
70
71
72
73
74
def __init__(
    self,
    redis_client: redis.Redis,
    key_prefix: str = "ratelimit:",
    use_redis_time: bool = False,
    fallback_storage: Optional[Storage] = None,
    fail_open: bool = False,
) -> None:

    self.redis = redis_client
    self.prefix = key_prefix
    self.use_redis_time = use_redis_time
    self.fallback_storage = fallback_storage
    self.fail_open = fail_open

pycurb.core.storage.redis_async.AsyncRedisStorage 

AsyncRedisStorage(
    redis_client,
    key_prefix="ratelimit:",
    use_redis_time=False,
    fallback_storage=None,
    fail_open=False,
)

Bases: AsyncStorage

Asynchronous redis storage for rate limiting.

Source code in src/pycurb/core/storage/redis_async.py 
62
63
64
65
66
67
68
69
70
71
72
73
74
def __init__(
    self,
    redis_client: aioredis.Redis,
    key_prefix: str = "ratelimit:",
    use_redis_time: bool = False,
    fallback_storage: Optional[AsyncStorage] = None,
    fail_open: bool = False,
) -> None:
    self.redis = redis_client
    self.prefix = key_prefix
    self.use_redis_time = use_redis_time
    self.fallback_storage = fallback_storage
    self.fail_open = fail_open

pycurb.core.limiter

pycurb.core.limiter.RateLimiter 

RateLimiter(storage, rules=None, resolver=None)

Core rate limiter engine.

Initialize the rate limiter

Parameters:

Name Type Description Default
storage Storage

A storage backend (e.g MemoryStorage, RedisStorage)

 required
rules Optional[List[LimitRule]]

A list of LimitRule objects.

 None
resolver Optional[Callable[[str], LimitRule]]

A custom rule resolver instance.

 None
Source code in src/pycurb/core/limiter.py 
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
def __init__(
    self,
    storage: Storage,
    rules: Optional[List[LimitRule]] = None,
    resolver: Optional[Callable[[str], LimitRule]] = None,
):
    """
    Initialize the rate limiter

    Args:
        storage: A storage backend (e.g MemoryStorage, RedisStorage)
        rules: A list of LimitRule objects.
        resolver: A custom rule resolver instance.
    """
    if rules is not None and resolver is not None:
        raise ValueError("Provide either 'rules' or 'resolver'. Not both.")

    self.storage = storage

    if resolver is not None:
        self.rule_resolver = resolver
    else:
        self.rule_resolver = RuleResolver(rules or [])

    self.algorithms: Dict[str, RateLimiterAlgorithm] = {
        "sliding_window": SlidingWindowAlgorithm(),
        "fixed_window": FixedWindowAlgorithm(),
        "token_bucket": TokenBucketAlgorithm(),
        "leaky_bucket": LeakyBucketAlgorithm(),
        "gcra": GcraAlgorithm(),
    }

Methods:

add_rule 

add_rule(rule)

Add or replace a rule

Source code in src/pycurb/core/limiter.py 
54
55
56
57
58
def add_rule(self, rule: LimitRule) -> None:
    """Add or replace a rule"""
    if not hasattr(self.rule_resolver, "add_rule"):
        raise TypeError("The rule resolver does not support dynamic rule addition")
    self.rule_resolver.add_rule(rule)  # type: ignore

check 

check(key, rule_names)

Check if a request with the given key is allowed under the named rule(s).

Parameters:

Name Type Description Default
key str

A unique identifier for the client (e.g IP, client_id, API key)

 required
rule_names Union[str, List[str]]

The name of the limit rule to apply. Also accepts multiple rules.

 required

Returns:

Name Type Description
RateLimitResult RateLimitResult

Decision and metadata

Raises:

Type Description
ValueError

If the rule name(s) is unknown or algorithm is not supported.
Source code in src/pycurb/core/limiter.py 
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
def check(self, key: str, rule_names: Union[str, List[str]]) -> RateLimitResult:
    """
    Check if a request with the given key is allowed under the named rule(s).

    Args:
        key: A unique identifier for the client (e.g IP, client_id, API key)
        rule_names: The name of the limit rule to apply. Also accepts multiple rules.

    Returns:
        RateLimitResult: Decision and metadata

    Raises:
        ValueError: If the rule name(s) is unknown or algorithm is not supported.
    """

    if isinstance(rule_names, str):
        return self._check_single(key, rule_names)
    results = []
    for name in rule_names:
        res = self._check_single(key, name)
        if not res.allowed:
            return res
        results.append(res)

    return min(results, key=lambda r: (r.remaining, r.reset_at))

remove_rule 

remove_rule(name)

Remove a rule by name

Source code in src/pycurb/core/limiter.py 
60
61
62
63
64
def remove_rule(self, name: str) -> None:
    """Remove a rule by name"""
    if not hasattr(self.rule_resolver, "remove_rule"):
        raise TypeError("The rule resolver does not support dynamic rule removal")
    self.rule_resolver.remove_rule(name)  # type: ignore

pycurb.core.limiter.RateLimiter.check 

check(key, rule_names)

Check if a request with the given key is allowed under the named rule(s).

Parameters:

Name Type Description Default
key str

A unique identifier for the client (e.g IP, client_id, API key)

 required
rule_names Union[str, List[str]]

The name of the limit rule to apply. Also accepts multiple rules.

 required

Returns:

Name Type Description
RateLimitResult RateLimitResult

Decision and metadata

Raises:

Type Description
ValueError

If the rule name(s) is unknown or algorithm is not supported.
Source code in src/pycurb/core/limiter.py 
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
def check(self, key: str, rule_names: Union[str, List[str]]) -> RateLimitResult:
    """
    Check if a request with the given key is allowed under the named rule(s).

    Args:
        key: A unique identifier for the client (e.g IP, client_id, API key)
        rule_names: The name of the limit rule to apply. Also accepts multiple rules.

    Returns:
        RateLimitResult: Decision and metadata

    Raises:
        ValueError: If the rule name(s) is unknown or algorithm is not supported.
    """

    if isinstance(rule_names, str):
        return self._check_single(key, rule_names)
    results = []
    for name in rule_names:
        res = self._check_single(key, name)
        if not res.allowed:
            return res
        results.append(res)

    return min(results, key=lambda r: (r.remaining, r.reset_at))

pycurb.core.limiter.RateLimiter.add_rule 

add_rule(rule)

Add or replace a rule

Source code in src/pycurb/core/limiter.py 
54
55
56
57
58
def add_rule(self, rule: LimitRule) -> None:
    """Add or replace a rule"""
    if not hasattr(self.rule_resolver, "add_rule"):
        raise TypeError("The rule resolver does not support dynamic rule addition")
    self.rule_resolver.add_rule(rule)  # type: ignore

pycurb.core.limiter.RateLimiter.remove_rule 

remove_rule(name)

Remove a rule by name

Source code in src/pycurb/core/limiter.py 
60
61
62
63
64
def remove_rule(self, name: str) -> None:
    """Remove a rule by name"""
    if not hasattr(self.rule_resolver, "remove_rule"):
        raise TypeError("The rule resolver does not support dynamic rule removal")
    self.rule_resolver.remove_rule(name)  # type: ignore

pycurb.core.limiter_async

pycurb.core.limiter_async.AsyncRateLimiter 

AsyncRateLimiter(storage, rules=None, resolver=None)

Async core rate limiter engine

Initialize the rate limiter

Parameters:

Name Type Description Default
storage AsyncStorage

A storage backend (e.g MemoryStorage, RedisStorage)

 required
rules Optional[List[LimitRule]]

A list of LimitRule objects

 None
resolver Optional[Callable[[str], LimitRule]]

A rule resolver instance.

 None
Source code in src/pycurb/core/limiter_async.py 
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
def __init__(
    self,
    storage: AsyncStorage,
    rules: Optional[List[LimitRule]] = None,
    resolver: Optional[Callable[[str], LimitRule]] = None,
):
    """
    Initialize the rate limiter

    Args:
        storage: A storage backend (e.g MemoryStorage, RedisStorage)
        rules: A list of LimitRule objects
        resolver: A rule resolver instance.
    """
    if rules is not None and resolver is not None:
        raise ValueError("Provide either 'rules' or 'resolver'. Not both.")
    self.storage = storage

    if resolver is not None:
        self.rule_resolver = resolver
    else:
        self.rule_resolver = AsyncRuleResolver(rules or [])

    self.algorithms: Dict[str, AsyncRateLimiterAlgorithm] = {
        "sliding_window": AsyncSlidingWindowAlgorithm(),
        "fixed_window": AsyncFixedWindowAlgorithm(),
        "token_bucket": AsyncTokenBucketAlgorithm(),
        "leaky_bucket": AsyncLeakyBucketAlgorithm(),
        "gcra": AsyncGcraAlgorithm(),
    }

Methods:

add_rule async 

add_rule(rule)

Add or replace a rule

Source code in src/pycurb/core/limiter_async.py 
56
57
58
59
60
async def add_rule(self, rule: LimitRule) -> None:
    """Add or replace a rule"""
    if not hasattr(self.rule_resolver, "add_rule"):
        raise TypeError("The rule resolver does not support dynamic rule addition")
    await self.rule_resolver.add_rule(rule)  # type: ignore

check async 

check(key, rule_names)

Check if a request with the given key is allowed under the named rule(s).

Parameters:

Name Type Description Default
key str

A unique identifier for the client (e.g IP, client_id, API key)

 required
rule_names Union[str, List[str]]

The name of the limit rule to apply. Also accepts multiple rules.

 required

Returns:

Name Type Description
RateLimitResult RateLimitResult

Decision and metadata

Raises:

Type Description
ValueError

If the rule name(s) is unknown or algorithm is not supported.
Source code in src/pycurb/core/limiter_async.py 
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
async def check(
    self, key: str, rule_names: Union[str, List[str]]
) -> RateLimitResult:
    """
    Check if a request with the given key is allowed under the named rule(s).

    Args:
        key: A unique identifier for the client (e.g IP, client_id, API key)
        rule_names: The name of the limit rule to apply. Also accepts multiple rules.

    Returns:
        RateLimitResult: Decision and metadata

    Raises:
        ValueError: If the rule name(s) is unknown or algorithm is not supported.
    """

    if isinstance(rule_names, str):
        return await self._check_single(key, rule_names)
    results = []
    for name in rule_names:
        res = await self._check_single(key, name)
        if not res.allowed:
            return res
        results.append(res)

    return min(results, key=lambda r: (r.remaining, r.reset_at))

remove_rule async 

remove_rule(name)

Remove a rule by name

Source code in src/pycurb/core/limiter_async.py 
62
63
64
65
66
async def remove_rule(self, name: str) -> None:
    """Remove a rule by name"""
    if not hasattr(self.rule_resolver, "remove_rule"):
        raise TypeError("The rule resolver does not support dynamic rule removal")
    await self.rule_resolver.remove_rule(name)  # type: ignore

pycurb.core.limiter_async.AsyncRateLimiter.check async 

check(key, rule_names)

Check if a request with the given key is allowed under the named rule(s).

Parameters:

Name Type Description Default
key str

A unique identifier for the client (e.g IP, client_id, API key)

 required
rule_names Union[str, List[str]]

The name of the limit rule to apply. Also accepts multiple rules.

 required

Returns:

Name Type Description
RateLimitResult RateLimitResult

Decision and metadata

Raises:

Type Description
ValueError

If the rule name(s) is unknown or algorithm is not supported.
Source code in src/pycurb/core/limiter_async.py 
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
async def check(
    self, key: str, rule_names: Union[str, List[str]]
) -> RateLimitResult:
    """
    Check if a request with the given key is allowed under the named rule(s).

    Args:
        key: A unique identifier for the client (e.g IP, client_id, API key)
        rule_names: The name of the limit rule to apply. Also accepts multiple rules.

    Returns:
        RateLimitResult: Decision and metadata

    Raises:
        ValueError: If the rule name(s) is unknown or algorithm is not supported.
    """

    if isinstance(rule_names, str):
        return await self._check_single(key, rule_names)
    results = []
    for name in rule_names:
        res = await self._check_single(key, name)
        if not res.allowed:
            return res
        results.append(res)

    return min(results, key=lambda r: (r.remaining, r.reset_at))

pycurb.core.resolver

pycurb.core.resolver.RuleResolver 

RuleResolver(initial_rules=None)

Bases: BaseRuleResolver

Sync rule resolver that stores rules in memory and allows dynamic addition/update. Implements the callable interface expected by RateLimiter.

Source code in src/pycurb/core/resolver.py 
75
76
77
78
79
80
81
def __init__(self, initial_rules: Optional[List[LimitRule]] = None):
    self._rules: Dict[str, LimitRule] = {}
    self._lock = threading.RLock()

    if initial_rules:
        for rule in initial_rules:
            self._rules[rule.name] = rule

Methods:

add_rule 

add_rule(rule)

Add or replace a rule by name.

Source code in src/pycurb/core/resolver.py 
83
84
85
86
def add_rule(self, rule: LimitRule) -> None:
    """Add or replace a rule by name."""
    with self._lock:
        self._rules[rule.name] = rule

remove_rule 

remove_rule(name)

Remove a rule.

Source code in src/pycurb/core/resolver.py 
88
89
90
91
def remove_rule(self, name: str) -> None:
    """Remove a rule."""
    with self._lock:
        self._rules.pop(name, None)

pycurb.core.resolver.AsyncRuleResolver 

AsyncRuleResolver(initial_rules=None)

Bases: AsyncBaseRuleResolver

Async rule resolver that stores rules in memory and allows dynamic addition/update. Implements the callable interface expected by RateLimiter.

Source code in src/pycurb/core/resolver.py 
44
45
46
47
48
49
50
def __init__(self, initial_rules: Optional[List[LimitRule]] = None):
    self._rules: Dict[str, LimitRule] = {}
    self._lock = asyncio.Lock()

    if initial_rules:
        for rule in initial_rules:
            self._rules[rule.name] = rule

Methods:

add_rule async 

add_rule(rule)

Add or replace a rule..

Source code in src/pycurb/core/resolver.py 
52
53
54
55
async def add_rule(self, rule: LimitRule) -> None:
    """Add or replace a rule.."""
    async with self._lock:
        self._rules[rule.name] = rule

remove_rule async 

remove_rule(name)

Remove a rule by name.

Source code in src/pycurb/core/resolver.py 
57
58
59
60
async def remove_rule(self, name: str) -> None:
    """Remove a rule by name."""
    async with self._lock:
        self._rules.pop(name, None)

pycurb.core.decorators

pycurb.core.decorators.rate_limit 

rate_limit(
    limiter,
    *,
    rule_name=None,
    limit_str=None,
    algorithm="sliding_window",
    key_extractor,
)

Unified decorator for rate limiting (both async and sync).

Parameters:

Name Type Description Default
limiter Union[RateLimiter, AsyncRateLimiter]

RateLimiter (async) or RateLimiterSync (sync) instance.

 required
rule_name Optional[Union[str, List[str]]]

Name of an existing rule in the limiter's resolver. Also accepts a list of names.

 None
limit_str Optional[str]

Shorthand string (e.g., "100/s") to create a new rule.

 None
algorithm str

Algorithm to use when creating a rule from limit_str.

 'sliding_window'
key_extractor Optional[Callable[..., str]]

Function extracting key (unique identifier) from function arguments.

 required

Raises:

Type Description
RateLimitExceeded

If rate limit has been exceeded.

Exactly one of rule_name or limit_str must be provided.

Source code in src/pycurb/core/decorators.py 
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
def rate_limit(
    limiter: Union[RateLimiter, AsyncRateLimiter],
    *,
    rule_name: Optional[Union[str, List[str]]] = None,
    limit_str: Optional[str] = None,
    algorithm: str = "sliding_window",
    key_extractor: Optional[Callable[..., str]],
):
    """
    Unified decorator for rate limiting (both async and sync).

    Args:
        limiter: RateLimiter (async) or RateLimiterSync (sync) instance.
        rule_name: Name of an existing rule in the limiter's resolver. Also accepts a list of names.
        limit_str: Shorthand string (e.g., "100/s") to create a new rule.
        algorithm: Algorithm to use when creating a rule from limit_str.
        key_extractor: Function extracting key (unique identifier) from function arguments.

    Raises:
        RateLimitExceeded: If rate limit has been exceeded.


    Exactly one of rule_name or limit_str must be provided.
    """
    if (rule_name is None) == (limit_str is None):
        raise ValueError("Provide exactly one of 'rule_name' or 'limit_str'")
    if key_extractor is None:
        raise TypeError("key_extractor must be provided")

    def decorator(func: Callable) -> Callable:
        func_is_async = inspect.iscoroutinefunction(func)
        limiter_is_async = inspect.iscoroutinefunction(limiter.check)

        # Validate consistency
        if func_is_async and not limiter_is_async:
            raise TypeError(
                "Async function requires an async RateLimiter (use RateLimiter, not RateLimiterSync)"
            )
        if not func_is_async and limiter_is_async:
            raise TypeError(
                "Sync function requires a sync RateLimiter (use RateLimiterSync, not RateLimiter)"
            )

        # Determine rule name and whether lazy creation is needed
        if rule_name is not None:
            effective_rule_name = rule_name
            need_rule_creation = False
        else:
            effective_rule_name = f"{func.__module__}.{func.__qualname__}"
            need_rule_creation = True
            _rule_created = False

        if func_is_async and limiter_is_async:
            limiter_async = cast(AsyncRateLimiter, limiter)

            @functools.wraps(func)
            async def async_wrapper(*args, **kwargs) -> Callable:
                nonlocal _rule_created
                if need_rule_creation and not _rule_created:
                    # Parse shorthand inside the wrapper (only once)
                    limit, window = parse_rate_limit_string(limit_str)  # type: ignore
                    rule = LimitRule(
                        name=cast(str, effective_rule_name),
                        algorithm=algorithm,  # type: ignore
                        limit=limit,
                        window=window,
                    )
                    resolver = limiter_async.rule_resolver
                    if not hasattr(resolver, "add_rule"):
                        raise TypeError(
                            "Resolver instance must have 'add_rule' attribute when using 'limit_str' in decorator"
                        )

                    await resolver.add_rule(rule)  # type: ignore
                    _rule_created = True

                key = key_extractor(*args, **kwargs)
                result = await limiter_async.check(key, effective_rule_name)
                if result.allowed:
                    return await func(*args, **kwargs)
                else:
                    raise RateLimitExceeded(result)

            wrapper = async_wrapper
        else:
            limiter_sync = cast(RateLimiter, limiter)

            @functools.wraps(func)
            def sync_wrapper(*args, **kwargs) -> Callable:
                nonlocal _rule_created
                if need_rule_creation and not _rule_created:
                    limit, window = parse_rate_limit_string(limit_str)  # type: ignore
                    rule = LimitRule(
                        name=cast(str, effective_rule_name),
                        algorithm=algorithm,  # type: ignore
                        limit=limit,
                        window=window,
                    )
                    resolver = limiter_sync.rule_resolver
                    if not hasattr(resolver, "add_rule"):
                        raise TypeError(
                            "Resolver class must have 'add_rule' attribute when using 'limit_str' in decorator"
                        )

                    resolver.add_rule(rule)  # type: ignore
                    _rule_created = True

                key = key_extractor(*args, **kwargs)
                result = limiter_sync.check(key, effective_rule_name)
                if result.allowed:
                    return func(*args, **kwargs)
                else:
                    raise RateLimitExceeded(result)

            wrapper = sync_wrapper
        return wrapper

    return decorator

pycurb.core.decorators.arg_extractor 

arg_extractor(*arg_names)

Helper to create a key extractor that combines values of specified argument names. Arguments must be passed as keyword arguments; positional arguments are ignored.

Example 
@rate_limit(limiter, rule_name="api", key_extractor=arg_extractor("user_id", "tenant"))
async def get_data(user_id: int, tenant: str):
Source code in src/pycurb/core/decorators.py 
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
def arg_extractor(*arg_names: str) -> Callable[..., str]:
    """
    Helper to create a key extractor that combines values of specified argument names.
    Arguments must be passed as keyword arguments; positional arguments are ignored.

    Example:
        ```python
        @rate_limit(limiter, rule_name="api", key_extractor=arg_extractor("user_id", "tenant"))
        async def get_data(user_id: int, tenant: str):
        ```
    """

    def extractor(*args: Any, **kwargs: Any) -> str:
        parts = [str(kwargs.get(name, "")) for name in arg_names]
        # fallback if all parts empty
        if not any(parts):
            raise ValueError(
                f"None of the specified argument names {arg_names} were found in kwargs: {kwargs}"
            )
        return ":".join(parts)

    return extractor

pycurb.utils

pycurb.utils.parse_rate_limit_string 

parse_rate_limit_string(rate_str)

Parse a shorthand rate limit string into (limit, window_seconds).

Supported formats (examples):

  • Single number (per-second): 100 -> limit=100, window=1
  • Per-unit shorthand: 100/s -> limit=100, window=1
  • Explicit duration with unit: 100/30s -> limit=100, window=30
  • Numeric window without unit: 100/60 -> limit=100, window=60
  • Larger units: 5/m -> limit=5, window=60; 10/h -> limit=10, window=3600

Examples::

```
100/s
100/30s
5/m
5/2m
10/h
10/2h
2/d
2/2d
100
100/60
```

Raises:

Type Description
ValueError

if format is invalid.
Source code in src/pycurb/utils.py 
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
def parse_rate_limit_string(rate_str: str) -> Tuple[int, int]:
    """
    Parse a shorthand rate limit string into (limit, window_seconds).

    Supported formats (examples):

    - Single number (per-second): ``100`` -> limit=100, window=1
    - Per-unit shorthand: ``100/s`` -> limit=100, window=1
    - Explicit duration with unit: ``100/30s`` -> limit=100, window=30
    - Numeric window without unit: ``100/60`` -> limit=100, window=60
    - Larger units: ``5/m`` -> limit=5, window=60; ``10/h`` -> limit=10, window=3600

    Examples::

        ```
        100/s
        100/30s
        5/m
        5/2m
        10/h
        10/2h
        2/d
        2/2d
        100
        100/60
        ```

    Raises:
        ValueError: if format is invalid.
    """
    rate_str = rate_str.strip().lower()
    if not rate_str:
        raise ValueError("Empty rate limit string")

    match = re.match(r"^(\d+)(?:/(\d*)([smhd]?))?$", rate_str)
    if not match:
        raise ValueError(f"Invalid rate limit format: '{rate_str}'")

    limit = int(match.group(1))
    window_part = match.group(2)  # may be empty string or None
    unit = match.group(3)  # may be empty string or None

    # Detect trailing slash: window_part is empty string, no unit, and there is a slash
    if "/" in rate_str and window_part == "" and not unit:
        raise ValueError(f"Invalid rate limit format: '{rate_str}'")

    # Determine window value
    if window_part is not None and window_part != "":
        window = int(window_part)
    else:
        # No explicit window number: use 1 as base
        window = 1

    # Apply unit multiplier
    if unit:
        multipliers = {"s": 1, "m": 60, "h": 3600, "d": 86400}
        if unit not in multipliers:
            raise ValueError(f"Unknown unit '{unit}'. Allowed: s, m, h, d")
        window *= multipliers[unit]

    if window <= 0:
        raise ValueError("Window must be positive")

    return limit, window

pycurb.utils.parse_duration 

parse_duration(duration_str)

Convert a shorthand duration string into seconds. Duration must be an integer followed by a unit. Supported units: - s: seconds - m: minutes - h: hours - d: days

Examples:

parse_duration("30s") -> 30 parse_duration("5m") -> 300 parse_duration("2h") -> 7200 parse_duration("1d") -> 86400

Raises:

Type Description
ValueError

If the format is invalid or unit is unsupported.
Source code in src/pycurb/utils.py 
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
def parse_duration(duration_str: str) -> int:
    """
    Convert a shorthand duration string into seconds.
    Duration must be an integer followed by a unit.
    Supported units:
        - s: seconds
        - m: minutes
        - h: hours
        - d: days

    Examples:
        parse_duration("30s") -> 30
        parse_duration("5m") -> 300
        parse_duration("2h") -> 7200
        parse_duration("1d") -> 86400

    Raises:
        ValueError: If the format is invalid or unit is unsupported.
    """
    duration_str = duration_str.strip().lower()
    if not duration_str:
        raise ValueError("Duration string cannot be empty.")

    unit = duration_str[-1]
    if unit not in ["s", "m", "h", "d"]:
        raise ValueError(
            f"Unsupported duration unit '{unit}'. Use 's', 'm', 'h', or 'd'."
        )

    try:
        value = int(duration_str[:-1])
    except ValueError:
        raise ValueError(
            f"Invalid duration value in '{duration_str}'. Must be an integer followed by a unit."
        )

    if value <= 0:
        raise ValueError("Duration value must be positive.")

    multipliers = {"s": 1, "m": 60, "h": 3600, "d": 86400}
    return value * multipliers[unit]