Metadata-Version: 2.4
Name: fastapi-limiter-valkey
Version: 0.2.0
Summary: A request rate limiter for fastapi, backed by Valkey
Keywords: fastapi,limiter,valkey
Author: Scraiber
Author-email: Scraiber <contact@scraiber.com>
License-Expression: Apache-2.0
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Requires-Dist: fastapi
Requires-Dist: fastapi-limiter>=0.2.0
Requires-Dist: pyrate-limiter>=4.0.0
Requires-Dist: valkey>=6.1.1
Requires-Python: >=3.10, <4
Project-URL: Homepage, https://github.com/scraiber/fastapi-limiter-valkey
Project-URL: Repository, https://github.com/scraiber/fastapi-limiter-valkey.git
Project-URL: Documentation, https://github.com/scraiber/fastapi-limiter-valkey
Description-Content-Type: text/markdown

# fastapi-limiter-valkey

[![pypi](https://img.shields.io/pypi/v/fastapi-limiter-valkey.svg?style=flat)](https://pypi.python.org/pypi/fastapi-limiter-valkey)
[![license](https://img.shields.io/github/license/scraiber/fastapi-limiter-valkey)](https://github.com/scraiber/fastapi-limiter-valkey/blob/main/LICENSE)
[![workflows](https://github.com/scraiber/fastapi-limiter-valkey/workflows/pypi/badge.svg)](https://github.com/scraiber/fastapi-limiter-valkey/actions?query=workflow:pypi)
[![workflows](https://github.com/scraiber/fastapi-limiter-valkey/workflows/ci/badge.svg)](https://github.com/scraiber/fastapi-limiter-valkey/actions?query=workflow:ci)

## Introduction

FastAPI-Limiter-Valkey is a rate limiting tool for [fastapi](https://github.com/tiangolo/fastapi)
routes, backed by [Valkey](https://valkey.io/).

It is a friendly fork of [FastAPI-Limiter](https://github.com/long2ice/fastapi-limiter), which since
`0.2.0` is powered by [pyrate-limiter](https://github.com/vutran1710/PyrateLimiter) and is fully
storage-agnostic. This package adds the missing piece for Valkey users: a `ValkeyBucket` — a
pyrate-limiter bucket typed for the [`valkey`](https://pypi.org/project/valkey/) client (the Redis
protocol works unchanged, so it runs on a Valkey server) — and re-exports the upstream `RateLimiter`
dependencies so you only need a single install.

> Since pyrate-limiter's `RedisBucket` only imports `redis` under `TYPE_CHECKING`, a Valkey client
> already works at runtime. `ValkeyBucket` exists so that type checkers are happy and the intent is
> explicit — you never install `redis`.

## Install

```shell script
> pip install fastapi-limiter-valkey
```

## Quick Start

Build a `Limiter` from a `ValkeyBucket` and pass it to the `RateLimiter` dependency. Because creating
a Valkey-backed bucket is async (it loads a Lua script onto the server), build the limiters in the
app's lifespan:

```py
from contextlib import asynccontextmanager

import uvicorn
import valkey.asyncio as valkey
from fastapi import Depends, FastAPI, Request, Response
from pyrate_limiter import Duration, Limiter, Rate

from fastapi_limiter_valkey import RateLimiter, ValkeyBucket

limiters: dict[str, RateLimiter] = {}


@asynccontextmanager
async def lifespan(_: FastAPI):
    client = valkey.from_url("valkey://localhost:6379", encoding="utf-8")
    bucket = await ValkeyBucket.init([Rate(2, Duration.SECOND * 5)], client, "index")
    limiters["index"] = RateLimiter(limiter=Limiter(bucket))
    yield
    await client.aclose()


app = FastAPI(lifespan=lifespan)


def rate_limit(key: str):
    async def dependency(request: Request, response: Response):
        return await limiters[key](request, response)

    return dependency


@app.get("/", dependencies=[Depends(rate_limit("index"))])
async def index():
    return {"msg": "Hello World"}


if __name__ == "__main__":
    uvicorn.run("main:app", reload=True)
```

See [`examples/main.py`](examples/main.py) for multiple limiters, websockets and the skip decorator.

## Usage

### ValkeyBucket

`ValkeyBucket.init(rates, client, bucket_key)` is `pyrate_limiter.RedisBucket.init` typed for a Valkey
client. It accepts both sync and async clients; with an async client it returns an awaitable that
resolves to the bucket. Wrap the result in a `pyrate_limiter.Limiter` and hand that to `RateLimiter`.

### RateLimiter

`RateLimiter` (re-exported from upstream `fastapi-limiter`) accepts:

- `limiter`: a `pyrate_limiter.Limiter` instance that defines the rate limiting rules.
- `identifier`: a callable to identify the request source, default is by IP + path.
- `callback`: a callable invoked when the rate limit is exceeded, default raises `HTTPException` with
  `429` status code.
- `blocking`: whether to block the request when the rate limit is exceeded, default is `False`.

### identifier

Default is `ip + path`; override it for e.g. `userid`:

```py
async def default_identifier(request: Union[Request, WebSocket]):
    forwarded = request.headers.get("X-Forwarded-For")
    if forwarded:
        ip = forwarded.split(",")[0]
    elif request.client:
        ip = request.client.host
    else:
        ip = "127.0.0.1"
    return ip + ":" + request.scope["path"]
```

### callback

Callback when the rate limit is exceeded, default raises `HTTPException` with `429`:

```py
def default_callback(*args, **kwargs):
    raise HTTPException(
        HTTP_429_TOO_MANY_REQUESTS,
        "Too Many Requests",
    )
```

## Multiple limiters

You can use multiple limiters in one route. Keep the stricter limiter (lower `seconds/times` ratio)
first.

```py
@app.get(
    "/multiple",
    dependencies=[Depends(rate_limit("multiple_short")), Depends(rate_limit("multiple_long"))],
)
async def multiple():
    return {"msg": "Hello World"}
```

## Skip rate limiting

Use the `skip_limiter` decorator to skip rate limiting for a specific route:

```py
from fastapi_limiter_valkey import skip_limiter

@app.get("/skip", dependencies=[Depends(rate_limit("skip"))])
@skip_limiter
async def skip_route():
    return {"msg": "This route skips rate limiting"}
```

## Rate limiting within a websocket

```py
from fastapi_limiter_valkey import WebSocketRateLimiter

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
    ratelimit = ws_limiters["ws"]  # a WebSocketRateLimiter built in the lifespan
    while True:
        try:
            data = await websocket.receive_text()
            await ratelimit(websocket, context_key=data)  # NB: context_key is optional
            await websocket.send_text("Hello, world")
        except HTTPException:
            await websocket.send_text("Hello again")
```

## License

This project is licensed under the
[Apache-2.0](https://github.com/scraiber/fastapi-limiter-valkey/blob/main/LICENSE) License.
