FastAPI Integration
VigilCV provides a sub-2ms pre-flight gate for FastAPI inference endpoints, intercepting corrupted or degraded images before they reach the model.
Full Example
import io
from fastapi import FastAPI, File, HTTPException, UploadFile
from fastapi.responses import JSONResponse
from vigilcv import VisionSentinel
from vigilcv.exceptions import CorruptImageError, QualityThresholdExceeded
app = FastAPI(title="VigilCV-Guarded Inference API")
Initialize once at startup — thread-safe and stateless
sentinel = VisionSentinel(
blur_threshold=100.0,
min_entropy=3.0,
max_underexposure_ratio=0.20,
max_overexposure_ratio=0.20,
raise_on_fail=True,
)
@app.post("/predict")
async def predict(file: UploadFile = File(...)) -> JSONResponse:
"""
Accept an image, run VigilCV pre-flight checks,
and forward to the ML model only if the image is pristine.
"""
image_bytes = await file.read()
try:
# Pre-flight guard: raises before model is invoked
metrics = sentinel.guard(io.BytesIO(image_bytes))
except CorruptImageError as e:
raise HTTPException(status_code=422, detail=f"Corrupt image: {e}")
except QualityThresholdExceeded as e:
raise HTTPException(status_code=422, detail=f"Quality check failed: {e}")
# Only pristine images reach here
predictions = my_model(image_bytes)
return JSONResponse({
"predictions": predictions,
"quality": {
"blur_score": round(metrics.blur_score, 2),
"entropy": round(metrics.shannon_entropy, 3),
"passed": True,
},
})
Startup Event (Warm Sentinel)
For high-throughput APIs, initialize the sentinel at startup:
from contextlib import asynccontextmanager
@asynccontextmanager
async def lifespan(app: FastAPI):
app.state.sentinel = VisionSentinel(blur_threshold=100.0)
yield
app = FastAPI(lifespan=lifespan)
Error Response Format
When VigilCV rejects an image, the API returns HTTP 422:
{
"detail": "Quality check failed: Laplacian variance 43.2 below threshold 100.0"
}
Performance
| Metric | Value |
|---|---|
| Pre-flight latency (512×512) | ~1.8ms |
| Memory overhead per request | ~0 MB (no model allocation) |
| Thread safety | ✅ Fully stateless |
| Async compatible | ✅ (sync I/O in thread pool) |