Create decorators that leverage FastAPI's `Depends()` and built-in dependencies, enabling you to inject dependencies directly into your decorators.
Project description
FastAPI decorators
Create decorators that leverage FastAPI's Depends()
and built-in dependencies, enabling you to inject dependencies directly into your decorators.
Installation
pip install fastapi-decorators
TL;DR
The library supplies the depends()
decorator function which effectively allows you to add argument dependencies to your FastAPI endpoints.
For example, the following three endpoints have the same signature:
# Using normal dependencies
@app.get("/items/{item_id}")
def read_item(item_id: int, _ = Depends(get_current_user)):
...
# Using add_dependency directly
@app.get("/items/{item_id}")
@depends(Depends(get_current_user))
def read_item(item_id: int):
...
# Using a custom decorator
def authorize():
def dependency(user = Depends(get_current_user)):
return user
return depends(Depends(dependency))
@app.get("/items/{item_id}")
@authorize()
def read_item(item_id: int):
...
Usage examples
- Using
depends()
directly - Logging decorator
- Authorization decorator
- Custom Response Header decorator
- Rate Limiting decorator
- Caching decorator
- Error Handling decorator
- Combining Multiple decorators
- Dependency injection with parameters
Using depends()
directly
If you prefer, you can use depends directly without creating a custom decorator:
from fastapi_decorators import depends
from fastapi import Depends, Header
async def verify_api_key(x_api_key: str = Header(...)):
if x_api_key != "expected-api-key":
raise HTTPException(status_code=403, detail="Forbidden")
@app.get("/secure-data")
@depends(Depends(verify_api_key))
def get_secure_data():
...
Logging decorator
Add a decorator to log incoming requests:
from fastapi_decorators import depends
from fastapi import Request, Depends
def log_request():
def dependency(request: Request):
print(f"Received request: {request.method} {request.url}")
return depends(Depends(dependency))
@app.get("/items/{item_id}")
@log_request()
def read_item(item_id: int):
...
Authorization decorator
Create a simple decorator that rejects unauthorized requests:
The API docs will reflect the authentication requirement for this endpoint because of the added OAuth2 dependency.
from fastapi_decorators import depends
from fastapi import Depends, HTTPException, Header
from fastapi.security import OAuth2PasswordBearer
from sqlalchemy.orm import Session
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
def authorize(*required_scopes: str):
def dependency(token: str = Depends(oauth2_scheme)):
# Your auth logic here
if not token:
raise HTTPException(status_code=401, detail="Unauthorized")
# Check scopes and permissions
pass
return depends(Depends(dependency))
@app.put("/users/{user_id}")
@authorize("users:write")
def update_user(*, user_id: int, user_update: UserUpdate):
...
Custom Response Header decorator
Create a decorator to add custom headers to responses:
from fastapi_decorators import depends
from fastapi import Response, Depends
def add_custom_header(name: str, value: str):
def dependency(response: Response):
response.headers[name] = value
return depends(Depends(dependency))
@app.get("/data")
@add_custom_header("X-Custom-Header", "MyValue")
def get_data():
...
Rate Limiting decorator
Add rate limiting to your endpoints:
from fastapi_decorators import depends
from fastapi import Depends, HTTPException, Request
from time import time
rate_limit_store = {}
def rate_limit(max_calls: int, period: int):
def dependency(ip_address: str = Depends(get_ip_address)):
# Simple rate limiting logic
now = time()
calls, last_reset = rate_limit_store.get(ip_address, (0, now))
if now - last_reset > period:
# Reset rate limit
calls = 0
last_reset = now
if calls >= max_calls:
raise HTTPException(status_code=429, detail="Too Many Requests")
calls += 1
rate_limit_store[ip_address] = (calls, last_reset)
return depends(Depends(dependency))
def get_ip_address(request: Request):
return request.client.host
@app.get("/limited-endpoint")
@rate_limit(max_calls=5, period=60)
def limited_endpoint():
...
Caching decorator
Add caching to your endpoints:
def get_cache() -> dict:
return {} # Use a real cache like Redis or Memcached
def cache_response(max_age: int = 5):
def decorator(func):
# Wrap the endpoint after adding the get_cache dependency
@depends(cache=Depends(get_cache))
@wraps(func)
def wrapper(*args, cache: dict, **kwargs):
key = func.__name__
if key in cache:
timestamp, data = cache[key]
if time() - timestamp < max_age:
# Cache hit
return data
# Cache miss - call the endpoint as usual
result = func(*args, **kwargs)
# Store the result in the cache
cache[key] = time(), result
return result
return wrapper
return decorator
@app.get("/cached-data")
@cache_response(max_age=10)
def get_cached_data():
...
Error Handling decorator
Create a decorator to handle exceptions and return custom responses:
from fastapi_decorators import depends
from fastapi import Depends, Response
def get_crash_log_storage() -> list:
return [] # Use a real storage like a database
def handle_errors():
def decorator(func):
# Wrap the endpoint after adding the crash_logs dependency
@depends(crash_logs = Depends(get_crash_log_storage))
@wraps(func)
def wrapper(*args, crash_logs: list, **kwargs):
try:
return func(*args, **kwargs)
except Exception as e:
# Log the error and return a custom response
crash_logs.append({ 'error': str(e), 'function': func.__name__ })
return JSONResponse(status_code=500, content={ "detail": str(e) })
return wrapper
return decorator
@app.get("/may-fail")
@handle_errors()
def may_fail_operation():
...
Combining Multiple decorators
You can combine multiple decorators to compose complex behavior:
@app.post("/submit")
@log_request()
@add_custom_header("X-Processed-By", "FastAPI")
@handle_errors()
def submit_data(data: DataModel):
...
Dependency injection with parameters
You can pass parameters to your dependencies through closures:
from fastapi_decorators import depends
from fastapi import Depends, HTTPException
def verify_role(required_role: str):
def dependency(current_user: User = Depends(get_current_user)):
if current_user.role != required_role:
raise HTTPException(status_code=403, detail="Forbidden")
return dependency
@app.get("/admin-area")
@depends(Depends(verify_role("admin")))
def admin_area():
...
Credits
Inspired by solutions suggested by @gocreating and @dmontagu.
Project details
Release history Release notifications | RSS feed
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.
Source Distribution
Built Distribution
Hashes for fastapi_decorators-1.0.4-py3-none-any.whl
Algorithm | Hash digest | |
---|---|---|
SHA256 | 335a1f13f5d64e87c4e11bb0f6eead1c0aa119d047ffab8a63ed35fc86530f50 |
|
MD5 | db622769078be4418ed97dc1e2eae2a0 |
|
BLAKE2b-256 | c9f314be87d2f13e50d24f21e1e96e39230342672941b344e30cae8cfb5f940a |