zanewalker
f3453cb0fc
- Refactor Redis backend connection handling and pool management - Update algorithm implementations with improved type annotations - Enhance config loader validation with stricter Pydantic schemas - Improve decorator and middleware error handling - Expand example scripts with better docstrings and usage patterns - Add new 00_basic_usage.py example for quick start - Reorganize examples directory structure - Fix type annotation inconsistencies across core modules - Update dependencies in pyproject.toml
474 lines
14 KiB
ReStructuredText
474 lines
14 KiB
ReStructuredText
Dependency Injection API
|
|
========================
|
|
|
|
If you're already using FastAPI's dependency injection system, you'll feel right
|
|
at home with ``RateLimitDependency``. It plugs directly into ``Depends``, giving
|
|
you rate limiting that works just like any other dependency—plus you get access
|
|
to rate limit info right inside your endpoint.
|
|
|
|
RateLimitDependency
|
|
-------------------
|
|
|
|
.. py:class:: RateLimitDependency(limit, window_size=60.0, *, algorithm=Algorithm.SLIDING_WINDOW_COUNTER, key_prefix="ratelimit", key_extractor=default_key_extractor, burst_size=None, error_message="Rate limit exceeded", status_code=429, skip_on_error=False, cost=1, exempt_when=None)
|
|
|
|
This is the main class you'll use for dependency-based rate limiting. Create
|
|
an instance, pass it to ``Depends()``, and you're done.
|
|
|
|
:param limit: Maximum number of requests allowed in the window.
|
|
:type limit: int
|
|
:param window_size: Time window in seconds. Defaults to 60.
|
|
:type window_size: float
|
|
:param algorithm: Rate limiting algorithm to use.
|
|
:type algorithm: Algorithm
|
|
:param key_prefix: Prefix for the rate limit key.
|
|
:type key_prefix: str
|
|
:param key_extractor: Function to extract client identifier from request.
|
|
:type key_extractor: Callable[[Request], str]
|
|
:param burst_size: Maximum burst size for token bucket/leaky bucket algorithms.
|
|
:type burst_size: int | None
|
|
:param error_message: Error message when rate limit is exceeded.
|
|
:type error_message: str
|
|
:param status_code: HTTP status code when rate limit is exceeded.
|
|
:type status_code: int
|
|
:param skip_on_error: Skip rate limiting if backend errors occur.
|
|
:type skip_on_error: bool
|
|
:param cost: Cost of each request (default 1).
|
|
:type cost: int
|
|
:param exempt_when: Function to determine if request should be exempt.
|
|
:type exempt_when: Callable[[Request], bool] | None
|
|
|
|
**Returns:** A ``RateLimitInfo`` object with details about the current rate limit state.
|
|
|
|
RateLimitInfo
|
|
-------------
|
|
|
|
When the dependency runs, it hands you back a ``RateLimitInfo`` object. Here's
|
|
what's inside:
|
|
|
|
.. py:class:: RateLimitInfo
|
|
|
|
:param limit: The configured request limit.
|
|
:type limit: int
|
|
:param remaining: Remaining requests in the current window.
|
|
:type remaining: int
|
|
:param reset_at: Unix timestamp when the window resets.
|
|
:type reset_at: float
|
|
:param retry_after: Seconds until retry is allowed (if rate limited).
|
|
:type retry_after: float | None
|
|
:param window_size: The configured window size in seconds.
|
|
:type window_size: float
|
|
|
|
.. py:method:: to_headers() -> dict[str, str]
|
|
|
|
Converts the rate limit info into standard HTTP headers. Handy if you want
|
|
to add these headers to your response manually.
|
|
|
|
:returns: A dictionary with ``X-RateLimit-Limit``, ``X-RateLimit-Remaining``,
|
|
``X-RateLimit-Reset``, and ``Retry-After`` (when applicable).
|
|
|
|
Setup
|
|
-----
|
|
|
|
Before you can use the dependency, you need to set up the rate limiter. The
|
|
cleanest way is with FastAPI's lifespan context manager:
|
|
|
|
.. code-block:: python
|
|
|
|
from contextlib import asynccontextmanager
|
|
from fastapi import FastAPI
|
|
from fastapi_traffic import MemoryBackend, RateLimiter
|
|
from fastapi_traffic.core.limiter import set_limiter
|
|
|
|
backend = MemoryBackend()
|
|
limiter = RateLimiter(backend)
|
|
|
|
@asynccontextmanager
|
|
async def lifespan(app: FastAPI):
|
|
await limiter.initialize()
|
|
set_limiter(limiter)
|
|
yield
|
|
await limiter.close()
|
|
|
|
app = FastAPI(lifespan=lifespan)
|
|
|
|
Basic Usage
|
|
-----------
|
|
|
|
Here's the simplest way to get started. Create a dependency instance and inject
|
|
it with ``Depends``:
|
|
|
|
.. code-block:: python
|
|
|
|
from fastapi import Depends, FastAPI, Request
|
|
from fastapi_traffic.core.decorator import RateLimitDependency
|
|
|
|
app = FastAPI()
|
|
|
|
# Create the rate limit dependency
|
|
rate_limit_dep = RateLimitDependency(limit=100, window_size=60)
|
|
|
|
@app.get("/api/data")
|
|
async def get_data(
|
|
request: Request,
|
|
rate_info=Depends(rate_limit_dep),
|
|
):
|
|
return {
|
|
"data": "here",
|
|
"remaining_requests": rate_info.remaining,
|
|
"reset_at": rate_info.reset_at,
|
|
}
|
|
|
|
Using Type Aliases
|
|
------------------
|
|
|
|
If you're using the same rate limit across multiple endpoints, type aliases
|
|
with ``Annotated`` make your code much cleaner:
|
|
|
|
.. code-block:: python
|
|
|
|
from typing import Annotated, TypeAlias
|
|
from fastapi import Depends, FastAPI, Request
|
|
from fastapi_traffic.core.decorator import RateLimitDependency
|
|
from fastapi_traffic.core.models import RateLimitInfo
|
|
|
|
app = FastAPI()
|
|
|
|
rate_limit_dep = RateLimitDependency(limit=100, window_size=60)
|
|
|
|
# Create a type alias for cleaner signatures
|
|
RateLimit: TypeAlias = Annotated[RateLimitInfo, Depends(rate_limit_dep)]
|
|
|
|
@app.get("/api/data")
|
|
async def get_data(request: Request, rate_info: RateLimit):
|
|
return {
|
|
"data": "here",
|
|
"remaining": rate_info.remaining,
|
|
}
|
|
|
|
Tiered Rate Limits
|
|
------------------
|
|
|
|
This is where dependency injection really shines. You can apply different rate
|
|
limits based on who's making the request—free users get 10 requests per minute,
|
|
pro users get 100, and enterprise gets 1000:
|
|
|
|
.. code-block:: python
|
|
|
|
from typing import Annotated, TypeAlias
|
|
from fastapi import Depends, FastAPI, Request
|
|
from fastapi_traffic.core.decorator import RateLimitDependency
|
|
from fastapi_traffic.core.models import RateLimitInfo
|
|
|
|
app = FastAPI()
|
|
|
|
# Define tier-specific limits
|
|
free_tier_limit = RateLimitDependency(
|
|
limit=10,
|
|
window_size=60,
|
|
key_prefix="free",
|
|
)
|
|
|
|
pro_tier_limit = RateLimitDependency(
|
|
limit=100,
|
|
window_size=60,
|
|
key_prefix="pro",
|
|
)
|
|
|
|
enterprise_tier_limit = RateLimitDependency(
|
|
limit=1000,
|
|
window_size=60,
|
|
key_prefix="enterprise",
|
|
)
|
|
|
|
def get_user_tier(request: Request) -> str:
|
|
"""Get user tier from header (in real app, from JWT/database)."""
|
|
return request.headers.get("X-User-Tier", "free")
|
|
|
|
TierDep: TypeAlias = Annotated[str, Depends(get_user_tier)]
|
|
|
|
async def tiered_rate_limit(
|
|
request: Request,
|
|
tier: TierDep,
|
|
) -> RateLimitInfo:
|
|
"""Apply different rate limits based on user tier."""
|
|
if tier == "enterprise":
|
|
return await enterprise_tier_limit(request)
|
|
elif tier == "pro":
|
|
return await pro_tier_limit(request)
|
|
else:
|
|
return await free_tier_limit(request)
|
|
|
|
TieredRateLimit: TypeAlias = Annotated[RateLimitInfo, Depends(tiered_rate_limit)]
|
|
|
|
@app.get("/api/resource")
|
|
async def get_resource(request: Request, rate_info: TieredRateLimit):
|
|
tier = get_user_tier(request)
|
|
return {
|
|
"tier": tier,
|
|
"remaining": rate_info.remaining,
|
|
"limit": rate_info.limit,
|
|
}
|
|
|
|
Custom Key Extraction
|
|
---------------------
|
|
|
|
By default, rate limits are tracked by IP address. But what if you want to rate
|
|
limit by API key instead? Just pass a custom ``key_extractor``:
|
|
|
|
.. code-block:: python
|
|
|
|
from fastapi import Depends, FastAPI, Request
|
|
from fastapi_traffic.core.decorator import RateLimitDependency
|
|
|
|
app = FastAPI()
|
|
|
|
def api_key_extractor(request: Request) -> str:
|
|
"""Extract API key for rate limiting."""
|
|
api_key = request.headers.get("X-API-Key", "anonymous")
|
|
return f"api:{api_key}"
|
|
|
|
api_rate_limit = RateLimitDependency(
|
|
limit=100,
|
|
window_size=3600, # 100 requests per hour
|
|
key_extractor=api_key_extractor,
|
|
)
|
|
|
|
@app.get("/api/resource")
|
|
async def api_resource(
|
|
request: Request,
|
|
rate_info=Depends(api_rate_limit),
|
|
):
|
|
return {
|
|
"data": "Resource data",
|
|
"requests_remaining": rate_info.remaining,
|
|
}
|
|
|
|
Multiple Rate Limits
|
|
--------------------
|
|
|
|
Sometimes you need layered protection—say, 10 requests per minute *and* 100
|
|
requests per hour. Dependencies make this easy to compose:
|
|
|
|
.. code-block:: python
|
|
|
|
from typing import Annotated, Any, TypeAlias
|
|
from fastapi import Depends, FastAPI, Request
|
|
from fastapi_traffic.core.decorator import RateLimitDependency
|
|
from fastapi_traffic.core.models import RateLimitInfo
|
|
|
|
app = FastAPI()
|
|
|
|
per_minute_limit = RateLimitDependency(
|
|
limit=10,
|
|
window_size=60,
|
|
key_prefix="minute",
|
|
)
|
|
|
|
per_hour_limit = RateLimitDependency(
|
|
limit=100,
|
|
window_size=3600,
|
|
key_prefix="hour",
|
|
)
|
|
|
|
PerMinuteLimit: TypeAlias = Annotated[RateLimitInfo, Depends(per_minute_limit)]
|
|
PerHourLimit: TypeAlias = Annotated[RateLimitInfo, Depends(per_hour_limit)]
|
|
|
|
async def combined_rate_limit(
|
|
request: Request,
|
|
minute_info: PerMinuteLimit,
|
|
hour_info: PerHourLimit,
|
|
) -> dict[str, Any]:
|
|
"""Apply both per-minute and per-hour limits."""
|
|
return {
|
|
"minute": {
|
|
"limit": minute_info.limit,
|
|
"remaining": minute_info.remaining,
|
|
},
|
|
"hour": {
|
|
"limit": hour_info.limit,
|
|
"remaining": hour_info.remaining,
|
|
},
|
|
}
|
|
|
|
CombinedRateLimit: TypeAlias = Annotated[dict[str, Any], Depends(combined_rate_limit)]
|
|
|
|
@app.get("/api/combined")
|
|
async def combined_endpoint(
|
|
request: Request,
|
|
rate_info: CombinedRateLimit,
|
|
):
|
|
return {
|
|
"message": "Success",
|
|
"rate_limits": rate_info,
|
|
}
|
|
|
|
Exemption Logic
|
|
---------------
|
|
|
|
Need to let certain requests bypass rate limiting entirely? Maybe internal
|
|
services or admin users? Use the ``exempt_when`` parameter:
|
|
|
|
.. code-block:: python
|
|
|
|
from fastapi import Depends, FastAPI, Request
|
|
from fastapi_traffic.core.decorator import RateLimitDependency
|
|
|
|
app = FastAPI()
|
|
|
|
def is_internal_request(request: Request) -> bool:
|
|
"""Check if request is from internal service."""
|
|
internal_token = request.headers.get("X-Internal-Token")
|
|
return internal_token == "internal-secret-token"
|
|
|
|
internal_exempt_limit = RateLimitDependency(
|
|
limit=10,
|
|
window_size=60,
|
|
exempt_when=is_internal_request,
|
|
)
|
|
|
|
@app.get("/api/internal")
|
|
async def internal_endpoint(
|
|
request: Request,
|
|
rate_info=Depends(internal_exempt_limit),
|
|
):
|
|
is_internal = is_internal_request(request)
|
|
return {
|
|
"message": "Success",
|
|
"is_internal": is_internal,
|
|
"rate_limit": None if is_internal else {
|
|
"remaining": rate_info.remaining,
|
|
},
|
|
}
|
|
|
|
Exception Handling
|
|
------------------
|
|
|
|
When a request exceeds the rate limit, a ``RateLimitExceeded`` exception is
|
|
raised. You'll want to catch this and return a proper response:
|
|
|
|
.. code-block:: python
|
|
|
|
from fastapi import FastAPI, Request
|
|
from fastapi.responses import JSONResponse
|
|
from fastapi_traffic import RateLimitExceeded
|
|
|
|
app = FastAPI()
|
|
|
|
@app.exception_handler(RateLimitExceeded)
|
|
async def rate_limit_handler(
|
|
request: Request,
|
|
exc: RateLimitExceeded,
|
|
) -> JSONResponse:
|
|
return JSONResponse(
|
|
status_code=429,
|
|
content={
|
|
"error": "rate_limit_exceeded",
|
|
"message": exc.message,
|
|
"retry_after": exc.retry_after,
|
|
},
|
|
)
|
|
|
|
Or if you prefer, there's a built-in helper that does the work for you:
|
|
|
|
.. code-block:: python
|
|
|
|
from fastapi import FastAPI, Request
|
|
from fastapi_traffic import RateLimitExceeded
|
|
from fastapi_traffic.core.decorator import create_rate_limit_response
|
|
|
|
app = FastAPI()
|
|
|
|
@app.exception_handler(RateLimitExceeded)
|
|
async def rate_limit_handler(request: Request, exc: RateLimitExceeded):
|
|
return create_rate_limit_response(exc, include_headers=True)
|
|
|
|
Complete Example
|
|
----------------
|
|
|
|
Here's everything put together in a working example you can copy and run:
|
|
|
|
.. code-block:: python
|
|
|
|
from contextlib import asynccontextmanager
|
|
from typing import Annotated, TypeAlias
|
|
|
|
from fastapi import Depends, FastAPI, Request
|
|
from fastapi.responses import JSONResponse
|
|
|
|
from fastapi_traffic import (
|
|
MemoryBackend,
|
|
RateLimiter,
|
|
RateLimitExceeded,
|
|
)
|
|
from fastapi_traffic.core.decorator import RateLimitDependency
|
|
from fastapi_traffic.core.limiter import set_limiter
|
|
from fastapi_traffic.core.models import RateLimitInfo
|
|
|
|
# Initialize backend and limiter
|
|
backend = MemoryBackend()
|
|
limiter = RateLimiter(backend)
|
|
|
|
@asynccontextmanager
|
|
async def lifespan(app: FastAPI):
|
|
await limiter.initialize()
|
|
set_limiter(limiter)
|
|
yield
|
|
await limiter.close()
|
|
|
|
app = FastAPI(lifespan=lifespan)
|
|
|
|
# Exception handler
|
|
@app.exception_handler(RateLimitExceeded)
|
|
async def rate_limit_handler(
|
|
request: Request,
|
|
exc: RateLimitExceeded,
|
|
) -> JSONResponse:
|
|
return JSONResponse(
|
|
status_code=429,
|
|
content={
|
|
"error": "rate_limit_exceeded",
|
|
"retry_after": exc.retry_after,
|
|
},
|
|
)
|
|
|
|
# Create dependency
|
|
api_rate_limit = RateLimitDependency(limit=100, window_size=60)
|
|
ApiRateLimit: TypeAlias = Annotated[RateLimitInfo, Depends(api_rate_limit)]
|
|
|
|
@app.get("/api/data")
|
|
async def get_data(request: Request, rate_info: ApiRateLimit):
|
|
return {
|
|
"data": "Your data here",
|
|
"rate_limit": {
|
|
"limit": rate_info.limit,
|
|
"remaining": rate_info.remaining,
|
|
"reset_at": rate_info.reset_at,
|
|
},
|
|
}
|
|
|
|
Decorator vs Dependency
|
|
-----------------------
|
|
|
|
Not sure which approach to use? Here's a quick guide:
|
|
|
|
**Go with the ``@rate_limit`` decorator if:**
|
|
|
|
- You just want to slap a rate limit on an endpoint and move on
|
|
- You don't care about the remaining request count inside your endpoint
|
|
- You're applying the same limit to a bunch of endpoints
|
|
|
|
**Go with ``RateLimitDependency`` if:**
|
|
|
|
- You want to show users how many requests they have left
|
|
- You need different limits for different user tiers
|
|
- You're stacking multiple rate limits (per-minute + per-hour)
|
|
- You're already using FastAPI's dependency system and want consistency
|
|
|
|
See Also
|
|
--------
|
|
|
|
- :doc:`decorator` - Decorator-based rate limiting
|
|
- :doc:`middleware` - Global middleware rate limiting
|
|
- :doc:`config` - Configuration options
|
|
- :doc:`exceptions` - Exception handling
|