Image moderation in Python with FastAPI
Last updated 2026-11-09
A FastAPI service that accepts images should decide before anything is written: is this file acceptable for the audience? This guide adds that decision to an upload endpoint with the official sfwapi package. You send the bytes, get a nudity level from 0 (Everyday) to 3 (Explicit) plus a blocked verdict against your policy, and map the API's errors onto HTTP responses your clients understand. The pattern is the same for Flask, Django or a background worker; only the framework glue changes.
Setup
pip install fastapi uvicorn python-multipart sfwapiCreate a key in the console and export it as SFWAPI_API_KEY. The free tier includes 500 credits a month, one credit per image, and no card is needed.
The endpoint
import os
from fastapi import FastAPI, File, HTTPException, UploadFile
from sfwapi import AsyncSfwApi, SfwApiError
app = FastAPI()
api = AsyncSfwApi(api_key=os.environ["SFWAPI_API_KEY"])
MAX_BYTES = 8 * 1024 * 1024 # free tier; paid accounts take 32 MB
POLICY = {"block_level": 2, "block_sex_objects": True}
@app.post("/images")
async def upload_image(file: UploadFile = File(...)):
if not (file.content_type or "").startswith("image/"):
raise HTTPException(415, "Please upload an image.")
data = await file.read()
if len(data) > MAX_BYTES:
raise HTTPException(413, "Images can be up to 8 MB.")
try:
r = await api.analyze_bytes(data, policy=POLICY)
except SfwApiError as e:
if e.status in (413, 415, 422):
raise HTTPException(422, "That file is not an image we can read.")
if e.status in (429, 503):
raise HTTPException(503, "Busy, try again in a moment.", headers={"Retry-After": "2"})
print("sfwapi", e.status, e.code, e.request_id)
raise HTTPException(502, "The image check is unavailable right now.")
if r["policy"]["blocked"]:
raise HTTPException(422, "This image cannot be uploaded here.")
scores = r["nudity"]["scores"].values()
needs_review = r["nudity"]["level"] >= 1 or any(35 < s < 65 for s in scores)
url = await store(data, file.content_type) # your storage layer
return {"url": url, "level": r["nudity"]["level"], "needs_review": needs_review}Run it with uvicorn main:app and post a file:
curl -F "file=@photo.jpg" http://localhost:8000/imagesAsyncSfwApi is built on httpx and can be shared across requests; create it once at import time or in a lifespan handler. The synchronous SfwApi has the same methods without await for scripts and workers.
Choosing the policy
A policy turns scores into a decision on the API side, so the service does not carry its own thresholds. block_level is the main knob:
1blocks swimwear and lingerie as well: strict, for communities with minors or professional spaces.2blocks implied nudity and explicit content: the default most products pick, bare skin is fine, nudity is not.3blocks explicit content only.
block_sex_objects: True adds adult toys and objects to the blocked set. If your users post drawings or anime, the art option says how illustrated content is judged (same_as_photos, allow, or block_from_1 to block_from_3). The policies page has a table of combinations that work for dating apps, general communities and adult platforms.
Reading the response
The response is a plain dict:
r["nudity"]["level"] # 0, 1, 2 or 3
r["nudity"]["scores"] # {"revealing": 96.2, "implied_nudity": 71.5, "explicit": 8.1}
r["flags"]["art"]["detected"] # drawing, anime, statue, stylised 3D
r["flags"]["sex_object"]["detected"]
r["policy"] # {"blocked": True, "reason": "level", "matched_level": 2}
r["usage"] # {"credits": 1, "balance": 499}Scores are calibrated so that 50 is the decision threshold on every level: a revealing of 51 and an explicit of 51 are equally "just over the line". Values between roughly 35 and 65 are the grey zone, which is what needs_review captures above. Pose, camera angle, framing and cleavage never raise the level on their own, so a level 2 or 3 is something you can act on. The levels page has the full definitions.
Mapping errors for your clients
The API uses one error envelope (error.type, error.code, error.message, error.doc_url, error.request_id) and the package raises SfwApiError with status, code, message and request_id. Group them by who can fix the problem:
- The user:
413 media_too_large,415 unsupported_media,422 decode_failed. Answer 4xx with a plain sentence. - Your account:
401 invalid_api_key,402 insufficient_credits,403 key_revoked. Alert yourself; decide whether uploads fail open or closed while the balance is empty. Auto-recharge in the console avoids the question. - Load:
429 rate_limitedorconcurrency_exceeded(free tier: 1 request per second, burst 3, 2 in flight; paid: 3 per second, burst 15, 8 in flight),503 overloaded. The package retries 429 and 503 twice withRetry-Afterfor URL requests; uploads are not retried, so pass the wait on to the client as the example does. - Transient:
502 upstream_internal,504 timeout. Retry once, with a smaller file on 504.
Failed requests are never charged. Log status, code and request_id, never the bytes or the file name. The errors page lists every code with the retry rule.
Checking an image by URL
When the image already has a public address, let the API download it instead of proxying the bytes through your service:
r = await api.analyze_url("https://cdn.example.com/u/123/avatar.jpg", policy=POLICY)Redirects are followed up to three hops; private, loopback and link-local targets answer 400 url_not_allowed. This is the right call for Telegram and Discord bots too, where the platform hands you a URL. The Telegram use case uses the same client.
Without the package
The request is a multipart POST with the policy as a JSON form field:
import json, os
import httpx
async def analyze(data: bytes, content_type: str) -> dict:
async with httpx.AsyncClient(timeout=70) as client:
res = await client.post(
"https://api.sfwapi.dev/v1/analyze",
headers={"Authorization": f"Bearer {os.environ['SFWAPI_API_KEY']}"},
files={"file": ("upload", data, content_type)},
data={"policy": json.dumps({"block_level": 2, "block_sex_objects": True})},
)
body = res.json()
if res.status_code >= 400:
raise RuntimeError(f"{body['error']['code']}: {body['error']['message']}")
return bodySynchronous requests time out after 60 seconds on the API side, so give the client a little more than that. The file name you send is discarded.
GIFs, videos and cost
GIFs work with the same call; frames are sampled evenly across the clip (5 by default, up to 32 with max_frames) and each analysed frame costs one credit. Videos need a paid account and can be sent by URL up to 64 MB. Pass include_frames=True to see which frame drove the level. The GIF and video page has guidance on how many frames to sample for which clip length.
One credit per image: 500 free checks a month, then $10 for 10,000 credits that never expire. Everything is on the pricing page.
Frequently asked questions
- Do I need the sfwapi package, or can I use requests or httpx directly?
- Either works. The package adds typed errors and retries on 429 and 503 that honour Retry-After. With httpx you send a multipart POST with the file and a policy field; the guide shows both.
- Is there a synchronous client?
- Yes. SfwApi has the same methods as AsyncSfwApi without await, for scripts, Celery workers and Django views.
- Which Python versions are supported?
- Python 3.9 and newer. The package is built on httpx and has no native dependencies.
- How large can an upload be?
- 8 MB on the free tier, 32 MB on paid accounts. Larger files answer 413 media_too_large and are not charged.
Related
- Moderating user uploads
- Nudity filter for Telegram bots
- Policies
- Errors and retries
- vs Google SafeSearch
Next steps
Read the quickstart to get your first result in two minutes, see pricing for the free tier and prepaid packs, or get a free API key and start with your own images.