
Research
Security News
The Growing Risk of Malicious Browser Extensions
Socket researchers uncover how browser extensions in trusted stores are used to hijack sessions, redirect traffic, and manipulate user behavior.
Supply Chain Security
Vulnerability
Quality
Maintenance
License
Create decorators with a @depends()
decorator that leverages FastAPI's Depends()
and built-in dependencies.
uv add fastapi-decorators
# or
pip install fastapi-decorators
The library supplies the depends()
decorator function which converts any function to a decorator that resolves FastAPI dependencies.
Create dependency-enabled decorators simply by using @depends
:
from fastapi_decorators import depends
from fastapi import Request, Depends, HTTPException
@depends
def authenticate(request: Request, db = Depends(get_db)):
if db.get_user(request.headers['x-user-id']) is None:
raise HTTPException(status_code=401, detail="Unauthenticated")
@app.get("/items/{item_id}")
@authenticate
def read_item(item_id: int):
...
You can use it to make decorator factories like so:
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
def authorize(scope: str):
@depends
def dependency(token: str = Depends(oauth2_scheme)):
if scope not in token.scopes:
raise HTTPException(status_code=403, detail="Unauthorized")
return dependency
# ... or in the explicit form:
def authorize(scope: str):
def dependency(token: str = Depends(oauth2_scheme)):
if scope not in token.scopes:
raise HTTPException(status_code=403, detail="Unauthorized")
return depends(dependency)
# Use it:
@app.put("/users/{user_id}")
@authorize("users:write")
def update_user(*, user_id: int, user_update: UserUpdate):
It can even be used to overwrite the endpoint logic while still using dependencies:
def cached():
def decorator(func):
@depends(cache=Depends(get_cache)) # Dependency `cache` registered here
@wraps(func)
def wrapper(*args, cache, **kwargs): # Dependency `cache` accessible here
if cache.hit():
return cache.get()
result = func(*args, **kwargs) # Execute the underlying endpoint function as normal
cache.set(result)
return result
return wrapper
return decorator
@app.get("/very-expensive-operation")
@cached()
def get_a_very_expensive_resource():
...
A more detailed version of the documentation is available here.
depends()
directlyIf 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():
...
The simplest way to create a decorator is to simply convert a function into a dependency decorator with @depends
:
from fastapi_decorators import depends
from fastapi import Request, Depends
@depends
def audit_request(request: Request, db: Session = Depends(get_db)):
log = AuditLog(request.host.ip, request.url.path)
db.add(log)
db.commit()
@app.get('/users/')
@audit_request
def get_users():
...
If preferred, the non-decorator syntax can also be used:
from fastapi_decorators import depends
from fastapi import Request, Depends
def audit_request():
def dependency(request: Request, db: Session = Depends(get_db)):
log = AuditLog(request.host.ip, request.url.path)
db.add(log)
db.commit()
return depends(dependency)
@app.get('/users/')
@audit_request()
def get_users():
...
In some cases, you need to provide variations of the same dependency. To do this, create a decorator factory:
def require_headers(*headers: str):
def dependency(request: Request):
if not all(header in request.headers for header in headers):
raise HTTPException(status=400, detail="Required headers not provided.")
return depends(dependency)
@app.put("/users/{user_id}")
@require_headers("X-API-KEY", "X-TENANT-ID")
def update_user(*, user_id: int, user_update: UserUpdate):
...
Add a decorator to log incoming requests:
from fastapi_decorators import depends
from fastapi import Request
@depends
def log_request(request: Request):
print(f"Received request: {request.method} {request.url}")
@app.get("/items/{item_id}")
@log_request
def read_item(item_id: int):
...
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", auto_error=False)
def authorize(*scopes: str):
def dependency(token: str | None = Depends(oauth2_scheme)):
if not token:
raise HTTPException(status_code=401, detail="Unauthenticated")
if not all(scope in token.scopes for scope in scopes):
raise HTTPException(status_code=403, detail="Unauthorized")
return depends(dependency)
@app.put("/users/{user_id}")
@authorize("users:read", "users:write")
def update_user(*, user_id: int, user_update: UserUpdate):
...
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(dependency)
@app.get("/data")
@add_custom_header("X-Custom-Header", "MyValue")
def get_data():
...
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(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():
...
Add caching to your endpoints:
cache_storage = {}
def get_cache() -> dict:
return cache_storage # 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)) # The `cache` dependency is defined here
@wraps(func)
def wrapper(*args, cache: dict, **kwargs): # The `cache` dependency can be accessed here
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():
...
get_cache
dependency with a keyword argument so we can access it in the wrapper function later.get_cache
dependency with the keyword argument cache
, we can access it in here. This also works for normal endpoints.Create a decorator to handle exceptions and return custom responses:
from fastapi_decorators import depends
from fastapi import Depends, Response
crash_logs = []
def get_crash_log_storage() -> list:
return crash_logs # 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():
...
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):
...
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(verify_role("admin"))
def admin_area():
...
If you're using a type checker like Mypy with a strict
configuration, the library exposes two useful types Decorator
and F
for satisfying type checks on decorators and their decorated functions:
from fastapi_decorators import depends
from fastapi_decorators.types import Decorator, F
def authorize() -> Decorator:
async def dependency(...) -> None:
...
return depends(dependency)
def cache_response(...) -> Decorator:
def decorator(func: F) -> F:
@depends(cache=Depends(get_cache))
@wraps(func)
def wrapper(...) -> Any:
...
return wrapper
return decorator
Inspired by solutions suggested by @gocreating and @dmontagu.
FAQs
Create decorators for your endpoints using FastAPI dependencies.
We found that fastapi-decorators demonstrated a healthy version release cadence and project activity because the last version was released less than a year ago. It has 1 open source maintainer collaborating on the project.
Did you know?
Socket for GitHub automatically highlights issues in each pull request and monitors the health of all your open source dependencies. Discover the contents of your packages and block harmful activity before you install or update your dependencies.
Research
Security News
Socket researchers uncover how browser extensions in trusted stores are used to hijack sessions, redirect traffic, and manipulate user behavior.
Research
Security News
An in-depth analysis of credential stealers, crypto drainers, cryptojackers, and clipboard hijackers abusing open source package registries to compromise Web3 development environments.
Security News
pnpm 10.12.1 introduces a global virtual store for faster installs and new options for managing dependencies with version catalogs.