FastAPI
Getting Started
If you like FastAPI as we do, you'll appreciate how naturally user.cleaning integrates with its async architecture. We'll use httpx, FastAPI's recommended HTTP client, to communicate with our API asynchronously.
First, install httpx in your project:
pip install httpxNext, store your API key securely in your environment variables. Never hardcode API keys directly in your source code:
# .env or environment
USER_CLEANING_API_KEY = "your-api-key"Next, store your API key securely in your environment variables and load it in your application:
import os
USER_CLEANING_API_KEY = os.environ.get("USER_CLEANING_API_KEY", "")If you don't have an API key yet, check out our Quickstart guide to generate one.
IP Forwarding for Rate Limiting
For user.cleaning's rate limiting to work correctly, you need to forward the real client IP with each request. Add this helper function to extract the client IP:
from fastapi import Request
def get_client_ip(request: Request) -> str:
x_forwarded_for = request.headers.get("X-Forwarded-For")
if x_forwarded_for:
return x_forwarded_for.split(",")[0].strip()
return request.headers.get("X-Real-IP") or request.client.host or "unknown"Production: If you're behind nginx, Cloudflare, or a load balancer, X-Forwarded-For is set automatically.
Development/testing: Without a reverse proxy forwarding X-Forwarded-For, the integration will see your server's IP instead of actual client IPs. This means all requests appear to come from the same IP, and after 5 disposable email attempts, all registrations will be blocked until the IP is unbanned.
If you experience unexpected registration blocks, verify your reverse proxy is forwarding client IPs correctly. You can unban blocked IPs at API Settings and configure number of registration attempts before block or disable it completely at Color Configuration.
Creating the Validation Function
Let's create an async helper function that communicates with the user.cleaning API:
import os
import httpx
async def check_email(email: str, client_ip: str) -> dict:
try:
async with httpx.AsyncClient() as client:
response = await client.post(
"https://api.user.cleaning/v1/external-api-requests/check-email",
params={"email": email},
headers={
"X-API-Key": USER_CLEANING_API_KEY,
"X-Forwarded-For": client_ip,
"X-Real-IP": client_ip,
},
timeout=5
)
if response.status_code == 403:
data = response.json()
return {"category": "banned", "message": data.get("detail", "Access denied")}
response.raise_for_status()
return response.json()
except httpx.RequestError:
return {"category": "white"}The API responds with a JSON object containing a category field that tells you how to handle the email address.
Integration Approaches
Validating in Route Handlers
The most straightforward approach is to call the validation function directly in your route handlers and raise an HTTPException when you encounter a disposable email:
from fastapi import HTTPException, Request, Form, status
@app.post("/register")
async def register(request: Request, email: str = Form(...), password: str = Form(...)):
client_ip = get_client_ip(request)
result = await check_email(email, client_ip)
if result["category"] == "banned":
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail=result["message"]
)
if result["category"] == "black":
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="Disposable emails are not allowed"
)
if result["category"] == "grey":
# Suspicious email — let them in but require phone verification
user = await create_user(email, password)
return {"message": "Please enter your phone number for verification", "user_id": user.id}
# 'white' — all good
user = await create_user(email, password)
return {"user_id": user.id}This approach works well when you have just a few routes that need email validation and want direct control over the logic.
Using FastAPI Dependencies
FastAPI's dependency injection system makes it easy to create reusable validation logic. By defining the validation as a dependency, you can apply it to any route that needs it:
from fastapi import Depends, HTTPException, Request, status
async def validate_email(request: Request, email: str) -> dict:
client_ip = get_client_ip(request)
result = await check_email(email, client_ip)
if result["category"] == "banned":
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail=result["message"]
)
if result["category"] == "black":
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="Disposable emails are not allowed"
)
return resultNow you can inject this validation into any route. The dependency will automatically run before your route handler executes, and you can access the validation result to determine if additional verification is needed:
@app.post("/register")
async def register(email: str, password: str, validation: dict = Depends(validate_email)):
requires_phone_verification = validation["category"] == "grey"
user = await create_user(email, password, requires_phone_verification=requires_phone_verification)
return {"user_id": user.id}Integrating with Pydantic Models
For maximum type safety and automatic validation, you can integrate the email check directly into your Pydantic models using field validators. This ensures that every time the model is instantiated, the email is automatically validated:
from pydantic import BaseModel, field_validator
import httpx
class RegistrationRequest(BaseModel):
email: str
password: str
@field_validator("email")
@classmethod
def validate_not_disposable(cls, v):
# Note: sync version for Pydantic validator
# IP forwarding not available in Pydantic validators
with httpx.Client() as client:
response = client.post(
"https://api.user.cleaning/v1/external-api-requests/check-email",
params={"email": v},
headers={"X-API-Key": os.environ["USER_CLEANING_API_KEY"]},
timeout=5.0
)
result = response.json()
if result["category"] == "black":
raise ValueError("Disposable emails are not allowed")
return vWith this approach, your route handlers become incredibly clean since all validation happens automatically when FastAPI parses the request body:
@app.post("/register")
async def register(data: RegistrationRequest):
# Email already validated by Pydantic
user = await create_user(data.email, data.password)
return {"user_id": user.id}Note that Pydantic validators run synchronously and don't have access to the request object, so IP forwarding isn't available with this approach. If you need IP-based rate limiting, use the route handler or dependency approach instead.
Understanding the Results
The category field can have three possible values:
- black — This is a disposable or temporary email address that should be blocked
- grey — Suspicious patterns were detected (like alias chains or plus addressing), consider requiring additional verification
- white — The email looks legitimate and can be accepted without restrictions
For a complete description of the output data structure and all available fields, see our Categories Reference.