Custom policies in FastAPI are essentially functions or classes that can be used to perform specific actions before or after a request is processed by an endpoint. These actions can include validating user input, checking user permissions, logging requests, and modifying responses. Custom policies can be applied globally to all endpoints in an application or selectively to specific endpoints.
FastAPI provides middleware as a way to perform actions on requests and responses globally. While middleware is useful for general-purpose tasks such as logging and error handling, custom policies are more focused on specific business rules and authorization logic. Custom policies can be more fine-grained and can be applied selectively to different parts of the application.
Let’s start by creating a simple custom policy that checks if a request contains a specific header.
from fastapi import FastAPI, Request, Depends
from typing import Callable
app = FastAPI()
def custom_header_policy(request: Request):
if 'X-Custom-Header' not in request.headers:
raise HTTPException(status_code=400, detail="Missing X-Custom-Header")
return True
@app.get("/protected")
async def protected_route(_: bool = Depends(custom_header_policy)):
return {"message": "This is a protected route"}
In this example, the custom_header_policy
function is a custom policy that checks if the X-Custom-Header
is present in the request headers. If the header is missing, it raises a HTTPException
with a status code of 400. The policy is then applied to the /protected
endpoint using the Depends
function.
To apply a custom policy globally to all endpoints in an application, you can use the dependency_overrides
feature of FastAPI.
from fastapi import FastAPI, Request, Depends
from typing import Callable
app = FastAPI()
def global_policy(request: Request):
# Add your global policy logic here
return True
app.dependency_overrides[Depends(lambda: True)] = Depends(global_policy)
@app.get("/")
async def root():
return {"message": "Hello, World!"}
In this example, the global_policy
function is applied globally to all endpoints in the application using the dependency_overrides
dictionary.
One of the most common use cases for custom policies is authentication and authorization. You can create custom policies to check if a user is authenticated and has the necessary permissions to access a particular endpoint.
from fastapi import FastAPI, Request, Depends, HTTPException
from typing import Callable
app = FastAPI()
def authenticated_user(request: Request):
# Implement your authentication logic here
if 'Authorization' not in request.headers:
raise HTTPException(status_code=401, detail="Unauthorized")
return True
def admin_user(request: Request):
# Implement your authorization logic here
if 'Admin' not in request.headers.get('Authorization', ''):
raise HTTPException(status_code=403, detail="Forbidden")
return True
@app.get("/admin")
async def admin_route(_: bool = Depends(authenticated_user), __: bool = Depends(admin_user)):
return {"message": "This is an admin-only route"}
In this example, the authenticated_user
policy checks if the user is authenticated, and the admin_user
policy checks if the user has admin privileges. Both policies are applied to the /admin
endpoint.
Custom policies can also be used to validate request and response data. You can create policies to check if the request data is in the correct format or if the response data meets certain criteria.
from fastapi import FastAPI, Request, Depends, HTTPException
from typing import Callable
app = FastAPI()
def request_validation_policy(request: Request):
try:
data = await request.json()
if 'name' not in data:
raise HTTPException(status_code=400, detail="Missing 'name' field in request")
return True
except ValueError:
raise HTTPException(status_code=400, detail="Invalid JSON data")
@app.post("/validate")
async def validate_route(_: bool = Depends(request_validation_policy)):
return {"message": "Request data is valid"}
In this example, the request_validation_policy
function validates the request data to ensure that it contains a name
field. If the data is invalid, it raises a HTTPException
with a status code of 400.
Each custom policy should have a single responsibility. Avoid creating policies that perform multiple unrelated tasks. This makes the policies easier to understand, test, and maintain.
FastAPI’s dependency injection system makes it easy to apply custom policies to endpoints. Use the Depends
function to inject policies into endpoints and make the code more modular.
Make sure to handle errors properly in your custom policies. Raise appropriate HTTPException
instances with clear error messages to provide useful feedback to the clients.
Write unit tests for your custom policies to ensure that they work as expected. Use testing frameworks such as pytest
to test the policies in isolation.
Custom policies in FastAPI provide a powerful way to extend the functionality of your API and enforce business rules and security measures. By understanding the fundamental concepts, usage methods, common practices, and best practices of custom policies, you can create more robust and secure FastAPI applications. Whether you need to implement authentication, authorization, or request/response validation, custom policies can help you achieve your goals in a flexible and modular way.