### Install FastAPI-Limiter

Source: https://github.com/long2ice/fastapi-limiter/blob/main/README.md

Install the package via pip.

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

--------------------------------

### Configure Global Rate Limiting Middleware

Source: https://github.com/long2ice/fastapi-limiter/blob/main/README.md

Apply rate limiting to all routes globally using RateLimiterMiddleware.

```python
from fastapi import FastAPI
from pyrate_limiter import Duration, Limiter, Rate

from fastapi_limiter.middleware import RateLimiterMiddleware

app = FastAPI()

app.add_middleware(
    RateLimiterMiddleware,
    limiter=Limiter(Rate(2, Duration.SECOND * 5)),
)


@app.get("/")
async def index():
    return {"msg": "Hello World"}
```

--------------------------------

### Apply Global Rate Limiting with Middleware

Source: https://context7.com/long2ice/fastapi-limiter/llms.txt

Use RateLimiterMiddleware to apply rate limiting across all routes globally, with optional skipping for specific paths.

```python
import uvicorn
from fastapi import FastAPI
from fastapi.requests import Request
from pyrate_limiter import Duration, Limiter, Rate

from fastapi_limiter.middleware import RateLimiterMiddleware

app = FastAPI()

# Skip rate limiting for health check endpoint
async def skip_health_check(request: Request):
    return request.scope["path"] == "/health"

# Global rate limiting: 2 requests per 5 seconds for all routes
app.add_middleware(
    RateLimiterMiddleware,
    limiter=Limiter(Rate(2, Duration.SECOND * 5)),
    skip=skip_health_check,
)

@app.get("/")
async def index():
    return {"msg": "Hello World"}

@app.get("/other")
async def other():
    return {"msg": "Other endpoint"}

@app.get("/health")
async def health():
    return {"status": "healthy"}  # Not rate limited

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

--------------------------------

### Implement Basic Rate Limiting

Source: https://github.com/long2ice/fastapi-limiter/blob/main/README.md

Use the RateLimiter dependency to restrict requests to a specific route.

```python
import uvicorn
from fastapi import Depends, FastAPI
from pyrate_limiter import Duration, Limiter, Rate

from fastapi_limiter.depends import RateLimiter

app = FastAPI()

@app.get(
    "/",
    dependencies=[Depends(RateLimiter(limiter=Limiter(Rate(2, Duration.SECOND * 5))))],
)
async def index():
    return {"msg": "Hello World"}


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

--------------------------------

### Apply Multiple Limiters

Source: https://github.com/long2ice/fastapi-limiter/blob/main/README.md

Stack multiple RateLimiter dependencies on a single route, keeping the strictest limiter first.

```python
@app.get(
    "/multiple",
    dependencies=[
        Depends(RateLimiter(limiter=Limiter(Rate(1, Duration.SECOND * 5)))),
        Depends(RateLimiter(limiter=Limiter(Rate(2, Duration.SECOND * 15)))),
    ],
)
async def multiple():
    return {"msg": "Hello World"}
```

--------------------------------

### Apply Multiple Rate Limiters

Source: https://context7.com/long2ice/fastapi-limiter/llms.txt

Enforce multiple time-window constraints on a single route by passing multiple dependencies.

```python
from fastapi import Depends, FastAPI
from pyrate_limiter import Duration, Limiter, Rate

from fastapi_limiter.depends import RateLimiter

app = FastAPI()

# Multiple limiters: 1 request per 5 seconds AND 2 requests per 15 seconds
@app.get(
    "/multiple",
    dependencies=[
        Depends(RateLimiter(limiter=Limiter(Rate(1, Duration.SECOND * 5)))),
        Depends(RateLimiter(limiter=Limiter(Rate(2, Duration.SECOND * 15)))),
    ],
)
async def multiple():
    return {"msg": "Hello World"}

# Behavior:
# - Can make 1 request every 5 seconds (burst limit)
# - Can only make 2 requests total in any 15 second window (sustained limit)
```

--------------------------------

### Implement Custom Callback

Source: https://context7.com/long2ice/fastapi-limiter/llms.txt

Define a callback function to handle rate limit violations, such as returning custom responses.

```python
from fastapi import Depends, FastAPI, HTTPException
from fastapi.responses import JSONResponse
from pyrate_limiter import Duration, Limiter, Rate
from starlette.requests import Request
from starlette.responses import Response

from fastapi_limiter.depends import RateLimiter

app = FastAPI()
```

--------------------------------

### Define Custom Rate Limit Callback

Source: https://context7.com/long2ice/fastapi-limiter/llms.txt

Implement a custom callback function to return a detailed JSON response when a rate limit is exceeded.

```python
async def custom_rate_limit_callback(request: Request, response: Response):
    raise HTTPException(
        status_code=429,
        detail={
            "error": "rate_limit_exceeded",
            "message": "You have exceeded the rate limit. Please try again later.",
            "path": request.url.path,
            "retry_after": 5
        },
        headers={"Retry-After": "5"}
    )

@app.get(
    "/api/limited",
    dependencies=[
        Depends(RateLimiter(
            limiter=Limiter(Rate(2, Duration.SECOND * 5)),
            callback=custom_rate_limit_callback
        ))
    ],
)
async def limited_endpoint():
    return {"status": "ok"}
```

--------------------------------

### Define Custom Callback

Source: https://github.com/long2ice/fastapi-limiter/blob/main/README.md

Define a custom action to perform when a rate limit is exceeded.

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

--------------------------------

### Implement WebSocket Rate Limiting

Source: https://context7.com/long2ice/fastapi-limiter/llms.txt

Apply rate limiting within WebSocket handlers to control message frequency per connection.

```python
from fastapi import FastAPI, HTTPException, WebSocket
from pyrate_limiter import Duration, Limiter, Rate
from starlette.websockets import WebSocketDisconnect

from fastapi_limiter.depends import WebSocketRateLimiter

app = FastAPI()

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
    # Create rate limiter: 1 message per 5 seconds
    ratelimit = WebSocketRateLimiter(limiter=Limiter(Rate(1, Duration.SECOND * 5)))

    while True:
        try:
            data = await websocket.receive_text()
            # Apply rate limiting to each message
            # context_key differentiates rate limits by message content (optional)
            await ratelimit(websocket, context_key=data)
            await websocket.send_text(f"Received: {data}")
        except WebSocketDisconnect:
            break
        except HTTPException:
            # Rate limit exceeded
            await websocket.send_text("Rate limit exceeded. Please slow down.")
```

--------------------------------

### Rate limiting within a websocket handler

Source: https://github.com/long2ice/fastapi-limiter/blob/main/README.md

Use WebSocketRateLimiter inside the websocket loop to control message frequency. The context_key parameter is optional and allows for granular limiting based on message content.

```python
from fastapi_limiter.depends import WebSocketRateLimiter

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
    ratelimit = WebSocketRateLimiter(limiter=Limiter(Rate(1, Duration.SECOND * 5)))
    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")
```

--------------------------------

### Default Identifier Implementation

Source: https://context7.com/long2ice/fastapi-limiter/llms.txt

The default identifier function combines the client's IP address and request path. It correctly handles proxied requests by inspecting the X-Forwarded-For header.

```python
from typing import Union
from starlette.requests import Request
from starlette.websockets import WebSocket

# This is the default identifier implementation
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"]
```

--------------------------------

### Conditionally Skip Rate Limiting

Source: https://github.com/long2ice/fastapi-limiter/blob/main/README.md

Provide a skip callable to conditionally bypass rate limiting for specific requests.

```python
from fastapi.requests import Request

async def skip(request: Request):
    return request.scope["path"] == "/skip"

@app.get(
    "/skip",
    dependencies=[
        Depends(RateLimiter(limiter=Limiter(Rate(1, Duration.SECOND * 5)), skip=skip))
    ],
)
async def skip_route():
    return {"msg": "This route skips rate limiting"}
```

--------------------------------

### Define Custom Identifiers

Source: https://context7.com/long2ice/fastapi-limiter/llms.txt

Override the default IP-based identification by providing a custom identifier function.

```python
from fastapi import Depends, FastAPI, Header
from pyrate_limiter import Duration, Limiter, Rate
from starlette.requests import Request

from fastapi_limiter.depends import RateLimiter

app = FastAPI()

# Custom identifier based on API key
async def api_key_identifier(request: Request):
    api_key = request.headers.get("X-API-Key", "anonymous")
    return f"{api_key}:{request.scope['path']}"

# Custom identifier based on user ID from header
async def user_identifier(request: Request):
    user_id = request.headers.get("X-User-ID", "anonymous")
    return f"user:{user_id}"

@app.get(
    "/api/data",
    dependencies=[
        Depends(RateLimiter(
            limiter=Limiter(Rate(10, Duration.MINUTE)),
            identifier=api_key_identifier
        ))
    ],
)
async def get_data():
    return {"data": "sensitive information"}

# Each unique API key gets its own rate limit bucket
# curl -H "X-API-Key: key123" http://localhost:8000/api/data
```

--------------------------------

### Skip Rate Limiting Conditionally

Source: https://context7.com/long2ice/fastapi-limiter/llms.txt

Use the skip parameter to bypass rate limiting for specific requests, such as those from admin users or internal services.

```python
from fastapi import Depends, FastAPI
from fastapi.requests import Request
from pyrate_limiter import Duration, Limiter, Rate

from fastapi_limiter.depends import RateLimiter

app = FastAPI()

# Skip rate limiting for admin users or internal requests
async def skip_for_admins(request: Request):
    # Skip if admin token is present
    if request.headers.get("X-Admin-Token") == "secret-admin-token":
        return True
    # Skip for internal service calls
    if request.headers.get("X-Internal-Service") == "true":
        return True
    return False

@app.get(
    "/api/resource",
    dependencies=[
        Depends(RateLimiter(
            limiter=Limiter(Rate(5, Duration.MINUTE)),
            skip=skip_for_admins
        ))
    ],
)
async def get_resource():
    return {"resource": "data"}
```

--------------------------------

### Define Custom Identifier

Source: https://github.com/long2ice/fastapi-limiter/blob/main/README.md

Override the default request identification logic by providing a custom callable.

```python
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"]
```

=== COMPLETE CONTENT === This response contains all available snippets from this library. No additional content exists. Do not make further requests.