Metadata-Version: 2.4
Name: fn-limit
Version: 0.1.0
Summary: A tiny, dependency-free, thread-safe and asyncio-safe rate-limiting decorator (token bucket / sliding window).
Author: NarayanTim
License: MIT
Project-URL: Homepage, https://github.com/NarayanTim/fn-limit
Project-URL: Issues, https://github.com/NarayanTim/fn-limit/issues
Keywords: rate-limit,throttle,token-bucket,sliding-window,decorator,asyncio
Classifier: Development Status :: 4 - Beta
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: MIT License
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.9
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Operating System :: OS Independent
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Classifier: Typing :: Typed
Requires-Python: >=3.10
Description-Content-Type: text/markdown
License-File: LICENSE
Provides-Extra: dev
Requires-Dist: pytest>=7.0; extra == "dev"
Requires-Dist: hypothesis>=6.0; extra == "dev"
Dynamic: license-file

# fn-limit

A small, dependency-free decorator for rate-limiting Python functions — sync or async,
single-process, thread-safe. Pick a strategy, a capacity, a period, and go.

```python
from fn_limit import rate_limit, RateLimitExceeded

@rate_limit(strategy="token_bucket", mode="raise", capacity=5, period=1.0)
def call_upstream_api():
    ...

call_upstream_api()  # fine, up to 5 times per second
call_upstream_api()  # 6th call within the window -> raises RateLimitExceeded
```

- **No dependencies.** Pure standard library (`threading`, `asyncio`, `time`, `collections`).
- **Sync and async, same decorator.** It detects `async def` automatically.
- **Thread-safe and asyncio-safe.** Internal locking handles concurrent callers correctly;
  see [Concurrency](#concurrency) for exactly what that guarantees.
- **Two strategies.** `token_bucket` (bursty, smooths out over time) and `sliding_window`
  (strict cap, no bursting). See [Strategies](#which-strategy-token_bucket-vs-sliding_window).
- **Two modes.** `raise` (reject over-limit calls) and `wait` (block until capacity frees up).

---

## Table of contents

- [Install](#install)
- [Quickstart](#quickstart)
- [Which strategy: token_bucket vs sliding_window](#which-strategy-token_bucket-vs-sliding_window)
- [Which mode: raise vs wait](#which-mode-raise-vs-wait)
- [Weighted calls](#weighted-calls)
- [Async support](#async-support)
- [Concurrency](#concurrency)
- [Lower-level API](#lower-level-api)
- [Examples](#examples)
  - [Basic: call-chain protection](#basic-call-chain-protection)
  - [API protection](#api-protection)
  - [Recursive / mutual call loops](#recursive--mutual-call-loops)
- [Framework examples](#framework-examples)
- [When *not* to use this](#when-not-to-use-this)
- [Gotchas](#gotchas)
- [How this compares to other options](#how-this-compares-to-other-options)
- [Testing this project](#testing-this-project)
- [FAQ](#faq)
- [License](#license)

---

## Install

```bash
pip install fn-limit
```

For running the test suite (not needed just to use the library):

```bash
pip install "fn-limit[dev]"   # adds pytest + hypothesis
```

Requires Python 3.10+. Zero runtime dependencies.

> Naming note: the **PyPI package** is `fn-limit` (hyphen, PyPI convention), the thing you
> `import` is `fn_limit` (underscore, required by Python syntax). Same for every module you
> see in this repo importing `from fn_limit import ...`.

---

## Quickstart

```python
from fn_limit import rate_limit, RateLimitExceeded

@rate_limit(strategy="token_bucket", mode="raise", capacity=5, period=1.0)
def ping():
    return "pong"

ping()  # "pong"
ping()  # "pong" ... up to 5 times in any 1-second window
ping()  # 6th call -> raises RateLimitExceeded
```

Catch it and read `retry_after`:

```python
try:
    ping()
except RateLimitExceeded as exc:
    print(f"try again in {exc.retry_after:.2f}s")
```

Or block instead of raising:

```python
@rate_limit(strategy="sliding_window", mode="wait", capacity=1, period=0.1)
def tick():
    return "tock"

tick()  # returns immediately
tick()  # blocks ~0.1s until the window frees up, then returns
```

**`period` is always seconds.** There's no separate minutes/hours mode — a "100 requests
per minute" budget is `capacity=100, period=60.0`, not `period=1.0`.

**`capacity` and `period` have no defaults on purpose.** There's no universally "right"
rate limit, so you're forced to pick one instead of silently inheriting a default that
doesn't match your use case.

---

## Which strategy: `token_bucket` vs `sliding_window`

|                                              | `token_bucket`                                                                                     | `sliding_window`                                                        |
| -------------------------------------------- | ---------------------------------------------------------------------------------------------------- | ------------------------------------------------------------------------- |
| Allows short bursts above the "average" rate | Yes, up to `capacity` at once, if idle beforehand                                                  | No, hard cap of `capacity` in any rolling `period`                      |
| Behavior after being idle                    | Refills up to full `capacity` — a burst is "available" again                                       | Old calls simply age out one at a time                                    |
| Memory / state                               | O(1) — just a float + a timestamp                                                                    | O(N) — a timestamp per call still inside the window                       |
| Good fit for                                 | Outbound API calls where the upstream tolerates bursts and you mostly care about the *average* rate | Enforcing a hard SLA / quota ("never more than N per minute, ever")       |
| Mental model                                 | A bucket that refills at a steady drip; each call drains it                                          | A log of recent timestamps; each call is only allowed if the log has room |

If you're not sure: **`token_bucket` is the default** and the more common choice for
outbound API calls, because most upstreams rate-limit you on an average-over-time basis and
tolerate short bursts fine. Reach for `sliding_window` when the limit is a hard promise
("we will never send more than 3 emails per user per hour", a billing quota, etc.) where a
burst right at a window boundary would be a real problem.

Concretely, the boundary-burst difference: with `token_bucket(capacity=5, period=1.0)`, if
you're idle for a while, you can spend all 5 tokens instantly, wait 0.99s, then spend 5 more
instantly — effectively 10 calls in ~1 second. With
`sliding_window(capacity=5, period=1.0)`, no 1-second window ever contains more than 5 calls,
full stop.

Full trade-off writeup, including the arithmetic, is in [DESIGN.md](DESIGN.md).

---

## Which mode: `raise` vs `wait`

|              | `mode="raise"`                                                                                    | `mode="wait"`                                                                                                   |
| ------------ | --------------------------------------------------------------------------------------------------- | ----------------------------------------------------------------------------------------------------------------- |
| Over limit   | Raises `RateLimitExceeded` immediately                                                             | Blocks the caller until capacity frees up                                                                         |
| Caller's job | Catch the exception, decide what to do (retry later, 429, fallback value)                           | Nothing — just wait, like a slower version of the call                                                            |
| Good fit for | Public-facing endpoints (turn it into an HTTP 429), places you want explicit backpressure signaling | Internal worker pools / batch jobs where "just go slower" beats "fail"                                            |
| Cost         | None while under the limit                                                                          | Every waiting thread (sync) or coroutine (async) is asleep, not spinning — cheap, but it does hold up that caller |

`RateLimitExceeded.retry_after` gives you a number of seconds you can use directly as an
HTTP `Retry-After` header, a backoff hint, etc.

---

## Weighted calls

Not every call is equal. `weight` lets a single call cost more than 1 unit of capacity:

```python
@rate_limit(strategy="token_bucket", mode="raise", capacity=10, period=1.0, weight=3)
def expensive_operation():
    ...
```

Now only 3 calls fit per window (`3 * 3 = 9 <= 10`, a 4th needs `12 > 10`).
`1 <= weight <= capacity` is checked once, at decoration time — a call that could never
succeed fails immediately instead of hanging forever in `mode="wait"`.

---

## Async support

Same decorator, same arguments — it detects `async def` via `inspect.iscoroutinefunction`
and switches internals automatically:

```python
@rate_limit(strategy="token_bucket", mode="wait", capacity=5, period=0.2)
async def fetch(url):
    ...
```

In `mode="wait"`, the async path uses `asyncio.sleep()` + `asyncio.Lock()`, not
`time.sleep()` + threads. `asyncio.sleep()` yields control back to the event loop while
waiting, so other tasks and requests keep making progress — a waiting coroutine does **not**
block the event loop the way `time.sleep()` would (see
`test_async_wait_mode_does_not_block_event_loop` in the test suite for an explicit proof of
this).

---

## Concurrency

Both algorithm classes (`TokenBucket`, `SlidingWindow`) hold a `threading.Lock` for the
short arithmetic of checking/consuming capacity — never across a sleep or I/O — so
`allow()` is atomic and safe to call from many threads or coroutines at once. Under
concurrent load, `mode="raise"` will admit *exactly* `capacity` callers and reject the rest,
with no over-admission race (there's a property/stress test locking this down:
`test_thread_safety_raise_mode_admits_exactly_capacity`).

`mode="wait"` additionally uses a `FairGate` (sync) or `asyncio.Lock` (async) so waiters are
served in the order they arrived, rather than every wakeup racing for newly-freed capacity.
The trade-off: only one waiter actively polls the limiter at a time. That's not a
throughput bottleneck (a poll is one cheap `allow()` call) — it's a fairness choice. See
`FairGate`'s docstring if you'd rather have a free-for-all instead.

**Important:** all of the above is about safety *within a single process*. See
[When not to use this](#when-not-to-use-this).

---

## Lower-level API

`TokenBucket` and `SlidingWindow` are available directly if you need to manage a limiter
instance yourself — for example, to share one budget across multiple functions, inspect
state, or integrate with a custom retry loop:

```python
from fn_limit.algo import TokenBucket, SlidingWindow

# Shared budget: both functions draw from the same 10/s pool
bucket = TokenBucket(capacity=10, period=0.1)

def read():
    if not bucket.allow():
        raise RuntimeError("rate limited")
    ...

def write():
    if not bucket.allow(weight=3):  # writes cost 3x
        raise RuntimeError("rate limited")
    ...
```

Key methods on both classes:

- `allow(weight=1) -> bool` — atomically consume capacity; returns `False` if unavailable.
- `wait_time(weight=1) -> float` — estimated seconds until `weight` capacity is free.
- `available -> int` — current free capacity (informational; see [Gotchas](#gotchas)).

These classes are considered part of the stable public API.

---

## Examples

Three standalone scripts in [`examples/`](examples/) show the most common
use cases from first principles. Each has a single `fn_limit` variable at
the top — change it and rerun to see the effect immediately.

### Basic: call-chain protection

**[`examples/basic_example.py`](examples/basic_example.py)**

The simplest possible demo. `function_c()` sits at the bottom of a
three-deep call chain and is the only thing decorated. Change `fn_limit`
and rerun to move the cutoff.

```
main()
  └─> function_a()
        └─> function_b()
              └─> function_c()   <- @rate_limit lives here
```

```python
from fn_limit import RateLimitExceeded, rate_limit

fn_limit = 1  # try 1, then 3

@rate_limit(strategy="token_bucket", mode="raise", capacity=fn_limit, period=10.0)
def function_c(call_number: int) -> str:
    return f"result-{call_number}"
```

Output with `fn_limit = 1`:

```
--- main() making call #1 ---
function_a(): calling function_b()
  function_b(): calling function_c()
    function_c(): doing the real work for call #1
  -> succeeded: result-1

--- main() making call #2 ---
function_a(): calling function_b()
  function_b(): calling function_c()
  -> BLOCKED: fn_limit reached (retry in 10.0s)
```

`function_c()` never has to know it is being called too many times — the
limiter is a separate concern from the actual work, and it fails loudly and
immediately rather than doing unbounded work silently.

---

### API protection

**[`examples/api_example.py`](examples/api_example.py)**

The recommended architecture for guarding any paid or rate-limited external
API. The limiter sits between your application logic and the function that
makes the real network call, so over-budget calls are rejected *before*
they reach the wire — no cost, no quota burn.

```
handle_user_request()        <- application layer
  └─> @rate_limit            <- budget checked here
        └─> ask_openai()     <- only reached if budget allows
              └─> OpenAI API (or simulated response)
```

`SIMULATE = True` by default — no network call, no API key, no cost. Flip
it to `False` only when you have `OPENAI_API_KEY` set.

```python
from fn_limit import RateLimitExceeded, rate_limit

SIMULATE = True
fn_limit = 1  # try 1, then 3, then 10

@rate_limit(strategy="token_bucket", mode="raise", capacity=fn_limit, period=60.0)
def ask_openai(prompt: str) -> str:
    if SIMULATE:
        return f"[simulated response to: {prompt!r}]"
    ...  # real OpenAI call
```

Output with `fn_limit = 1`:

```
[alice] OK: [simulated response to: 'question #1']
[alice] BLOCKED — retry in 60.0s
[alice] BLOCKED — retry in 60.0s
[alice] BLOCKED — retry in 60.0s
[alice] BLOCKED — retry in 60.0s
```

Output with `fn_limit = 3`:

```
[alice] OK: [simulated response to: 'question #1']
[alice] OK: [simulated response to: 'question #2']
[alice] OK: [simulated response to: 'question #3']
[alice] BLOCKED — retry in 60.0s
[alice] BLOCKED — retry in 60.0s
```

---

### Recursive / mutual call loops

**[`examples/recursive_example.py`](examples/recursive_example.py)**

Without a budget, a mutual A↔B loop — a missing base case, a mis-wired
chatbot tool-call, a retried webhook — runs until something external stops
it: your API bill, a provider hard-limit, or a crash. With `@rate_limit`
the chain stops itself cleanly, no hop-counting safety net required.

The file shows both sides: the unprotected version (capped at 20 hops so
the demo can't actually loop forever) and the protected version.

```
function_a_protected()
  └─> _guarded_hop()     <- @rate_limit on every hop
  └─> function_b_protected()
        └─> _guarded_hop()
        └─> function_a_protected()  ... until budget runs out
              └─> RateLimitExceeded raised, propagates up once
```

```python
from fn_limit import RateLimitExceeded, rate_limit

fn_limit = 3  # try 3, then 6

@rate_limit(strategy="token_bucket", mode="raise", capacity=fn_limit, period=5.0)
def _guarded_hop() -> None:
    """Every A↔B hop passes through here."""
```

Output with `fn_limit = 3`:

```
Without fn_limit (capped at 20 hops so this demo can't run forever):
ABABABABABABABABABABA

With fn_limit=3:
ABA
STOP: fn_limit reached (retry in 5.0s)
```

---

## Framework examples

Runnable, self-contained examples live in [`examples/`](examples/):

- [`examples/threading_example.py`](examples/threading_example.py) — a limiter shared across
  a pool of worker threads, both `mode="wait"` and `mode="raise"`.
- [`examples/flask_example.py`](examples/flask_example.py) — turning `RateLimitExceeded` into
  an HTTP 429 with a `Retry-After` header, plus a reusable decorator so you don't repeat the
  try/except in every view.
- [`examples/fastapi_example.py`](examples/fastapi_example.py) — async routes,
  `mode="wait"` without blocking the event loop, and a dependency-based variant.
- [`examples/django_example.py`](examples/django_example.py) — the same 429-conversion
  pattern as Flask, Django-style.

The short version, for a Flask view:

```python
from flask import jsonify
from fn_limit import rate_limit, RateLimitExceeded

@rate_limit(strategy="sliding_window", mode="raise", capacity=5, period=10.0)
def _get_data():
    return {"data": "here you go"}

@app.route("/api/data")
def get_data():
    try:
        return jsonify(_get_data())
    except RateLimitExceeded as exc:
        resp = jsonify({"error": "rate limit exceeded"})
        resp.status_code = 429
        resp.headers["Retry-After"] = str(round(exc.retry_after, 2))
        return resp
```

And an async FastAPI route that smooths bursts instead of rejecting:

```python
from fn_limit import rate_limit

@rate_limit(strategy="token_bucket", mode="wait", capacity=10, period=0.1)
async def _do_work():
    ...

@app.get("/fast")
async def fast_endpoint():
    return await _do_work()
```

---

## When *not* to use this

fn_limit is intentionally a single-process, in-memory library. That's the right tool for
"protect this process from doing too much work" — it is **not** the right tool for:

- **Rate limiting shared across multiple processes or machines** (e.g. `gunicorn -w 4`, or a
  fleet of pods). Each process gets its own independent limiter state; four workers each
  enforcing "5/second" gives you ~20/second in aggregate, not 5. For a limit that's actually
  global across processes, you need shared state — typically Redis (see
  [`limits`](https://pypi.org/project/limits/) or a Redis `INCR`+`EXPIRE`/Lua-script pattern).
- **Per-client / per-user / per-IP limiting.** fn_limit has no concept of a "key" — one
  `@rate_limit(...)` decorator is one shared budget for every caller of that function. If you
  want "5 requests per minute per API key," you need a dict of limiters keyed by that
  identifier yourself (see [Gotchas](#gotchas) below), or a framework-specific tool that
  already does this ([Django REST Framework throttles](https://www.django-rest-framework.org/api-guide/throttling/),
  [slowapi](https://pypi.org/project/slowapi/) for FastAPI).
- **Surviving a process restart.** State is in memory; restart the process and every limiter
  resets to full capacity.

If any of those apply, this library isn't a bad starting point conceptually (the
token-bucket/sliding-window math is the same either way), but you'll want a
Redis-backed limiter instead of this one.

---

## Gotchas

- **Reusing one decorator instance across multiple functions shares its limiter.**

  ```python
  limited = rate_limit(strategy="token_bucket", mode="raise", capacity=5, period=1.0)

  @limited
  def f(): ...

  @limited
  def g(): ...
  ```

  `f` and `g` share the *same* underlying bucket here, because `rate_limit(...)` builds the
  limiter once, when it's called, and `@limited` just applies that same closure twice. Calls
  to `f()` and `g()` draw from one combined budget of 5/second, not 5/second each. This is
  sometimes exactly what you want (one budget for "all database writes," say) — just know
  it's happening. If you want independent limits, call `rate_limit(...)` again for each
  function (or just use `@rate_limit(...)` directly above each one, which is the common case
  and does the right thing automatically).

- **`wait_time()` is an estimate, not a guarantee.**

  For `SlidingWindow`, `wait_time()` is computed from when the *oldest* logged entry ages
  out. If the requested `weight` needs more than that one entry to free up enough room, the
  real wait can be longer than what `wait_time()` reports — treat it as a lower bound.

  For `TokenBucket`, `wait_time()` can be *longer* than the estimate under contention:
  other concurrent callers may consume newly-refilled tokens before you do, pushing your
  actual wait out further.

  `mode="wait"` handles both cases correctly on its own (it loops and re-checks). But if
  you're reading `retry_after` / `wait_time()` directly to show a user a countdown, treat
  it as an optimistic estimate, not an exact figure.

- **`available` is informational, not a reservation.** Checking `tb.available` and then
  calling `tb.allow()` is two separate operations; another caller can consume capacity in
  between. Always use the return value of `allow()` (or catch `RateLimitExceeded`) to decide
  whether a call actually went through — don't gate on `available` and assume `allow()` will
  agree.

- **Per-key limiting needs a dict of limiters.**

  ```python
  from collections import defaultdict
  from fn_limit.algo import TokenBucket

  _limiters: dict[str, TokenBucket] = defaultdict(
      lambda: TokenBucket(capacity=5, period=1.0)
  )

  def call_for_user(user_id: str):
      if not _limiters[user_id].allow():
          raise RateLimitExceeded()
      ...
  ```

  This pattern works fine for a modest number of keys. For unbounded key spaces, add an
  eviction strategy (e.g. `cachetools.TTLCache`) so limiters for inactive keys don't
  accumulate indefinitely.

---

## How this compares to other options

A rough, non-exhaustive comparison to help you pick:

|                                | fn_limit                     | [`ratelimit`](https://pypi.org/project/ratelimit/) | [`limits`](https://pypi.org/project/limits/)              | [`slowapi`](https://pypi.org/project/slowapi/)  |
| ------------------------------ | ---------------------------- | --------------------------------------------------- | ---------------------------------------------------------- | ------------------------------------------------ |
| Scope                          | Single process               | Single process                                      | Single process **or** Redis/Memcached-backed               | FastAPI/Starlette middleware, in-memory or Redis |
| Async support                  | Yes, same decorator          | No                                                  | Partial, varies by backend                                 | Yes (it's async-first)                           |
| Per-key (per-user/IP) limiting | No, build it yourself        | No                                                  | Yes, via a "key" you pass in                               | Yes, built for this                              |
| Strategies                     | token bucket, sliding window | leaky-bucket style                                  | several (fixed window, sliding window, moving window, ...) | fixed/sliding window                             |
| Framework-specific             | No, it's a plain decorator   | No                                                  | No                                                         | FastAPI/Starlette only                           |

If you need per-client limiting or multi-process consistency, `limits` (with a Redis
backend) or a framework-native tool like `slowapi`/DRF throttles is a better fit than
fn_limit out of the box. fn_limit's niche is: a plain function decorator, zero
dependencies, sync and async, that you can drop onto *any* callable in *one* process.

---

## Testing this project

```bash
pip install -e ".[dev]"
pytest
```

The suite includes:

- Unit tests per component (`test_token_bucket.py`, `test_sliding_window.py`,
  `test_decorator.py`, `test_fair_gate.py`, `test_limiter_factory.py`, `test_validation.py`).
- Property-based tests (`test_property_based.py`, via Hypothesis) that generate random
  operation sequences instead of hand-picked scenarios, because the bugs that hit this kind
  of code (off-by-ones, stale state, boundary conditions) tend to show up on inputs a human
  wouldn't think to write by hand.
- A `fake_clock` fixture (`conftest.py`) that patches `time.monotonic`/`time.sleep` so tests
  that exercise refill/expiry/blocking behavior don't spend real wall-clock time waiting.
  This matters more than it sounds like: one single Hypothesis-driven test that used to call
  real `time.sleep()` took **105 seconds** on its own; converted to `fake_clock`, the entire
  95-test suite runs in **~2.6 seconds**. See [REVIEW.md](REVIEW.md) for the full before/after
  breakdown and which tests were deliberately *left* on real time (thread-scheduling tests —
  faking the clock doesn't make threads interleave any faster or more deterministically).

---

## FAQ

**Does this work with `multiprocessing`?**
No — each process has its own memory, so each process gets its own independent limiter. See
[When not to use this](#when-not-to-use-this).

**What happens if `capacity` or `period` are invalid?**
`ValueError`, raised immediately when you call `rate_limit(...)` (i.e. at decoration time,
not on the first call) — so a bad config fails fast at import time instead of surfacing
later as confusing runtime behavior.

**Can I change the limit after decorating a function?**
Not directly — `capacity`/`period` are fixed for the lifetime of the limiter created inside
`rate_limit(...)`. If you need this, decorate a thin wrapper and swap which `rate_limit(...)`
it delegates to, or use `TokenBucket`/`SlidingWindow` directly instead of the decorator (see
[Lower-level API](#lower-level-api)) and manage the instance yourself.

**Is it typed?**
Yes, fully annotated, no `Any` leakage in the public API beyond `*args`/`**kwargs` forwarding.

**What's the minimum Python version?**
Python 3.10. This is enforced at install time via `requires-python = ">=3.10"` in
`pyproject.toml`.

---

## License

MIT. See [LICENSE](LICENSE).
