File size: 5,287 Bytes
9b5157d | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 | """
Photo Verification API — FastAPI app factory.
"""
import logging
import os
import time
from contextlib import asynccontextmanager
from fastapi import FastAPI, Request
from fastapi.exceptions import RequestValidationError
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import JSONResponse, RedirectResponse
from fastapi.staticfiles import StaticFiles
from slowapi import _rate_limit_exceeded_handler
from slowapi.errors import RateLimitExceeded
from app.core.config import settings
from app.core.exceptions import AppError
from app.core.logging_config import configure_logging
from app.middleware.correlation_id import CorrelationIdMiddleware, get_request_id
from app.middleware.rate_limit import limiter
from app.middleware.timing import TimingMiddleware
from app.metrics.prometheus import setup_metrics
from app.routers import verification
from app.routers import health as health_router
from app.schemas.errors import ErrorResponse, ErrorDetail
# Silence TensorFlow noise before any TF import
os.environ.setdefault("TF_CPP_MIN_LOG_LEVEL", "3")
os.environ.setdefault("TF_ENABLE_ONEDNN_OPTS", "0")
configure_logging(settings.LOG_LEVEL, settings.LOG_FORMAT)
logger = logging.getLogger(__name__)
@asynccontextmanager
async def lifespan(app: FastAPI):
"""Load all CV models at startup; drain thread pool on shutdown."""
from app.services.face_analysis import face_analysis_service
from app.services.liveness import liveness_service
from app.core.dependencies import cv_executor
logger.info("starting up", extra={"version": settings.VERSION})
face_analysis_service.load()
liveness_service.load()
logger.info("all models loaded — API ready")
yield
logger.info("shutting down — draining CV thread pool")
cv_executor.shutdown(wait=True, cancel_futures=False)
logger.info("shutdown complete")
def _make_error_response(status_code: int, error_code: str, message: str, context: dict = {}) -> JSONResponse:
return JSONResponse(
status_code=status_code,
content=ErrorResponse(
error=ErrorDetail(
error_code=error_code,
message=message,
request_id=get_request_id(),
context=context,
)
).model_dump(),
)
def create_app() -> FastAPI:
app = FastAPI(
title=settings.APP_NAME,
version=settings.VERSION,
description="""
## Photo Verification API
Simulates the pose-challenge verification system used in apps like Bumble/Tinder.
### Pipeline
1. **Liveness detection** — MiniFASNet (Silent-Face-Anti-Spoofing) via DeepFace
2. **Face landmark extraction** — MediaPipe FaceLandmarker (478 3D points)
3. **Head pose estimation** — PnP solver (yaw/pitch/roll in degrees)
4. **Challenge matching** — threshold-based pose verification
### No training needed — all models are pretrained.
""",
contact={"name": "API Support"},
openapi_tags=[
{"name": "Verification v1", "description": "Pose challenge verification endpoints"},
{"name": "Health", "description": "Service health and readiness"},
],
docs_url="/docs" if settings.DEBUG else "/docs", # keep docs accessible for demo
redoc_url="/redoc" if settings.DEBUG else "/redoc",
lifespan=lifespan,
)
# --- Rate limiter state ---
app.state.limiter = limiter
# --- Middleware (last added = outermost on request) ---
app.add_middleware(CORSMiddleware, allow_origins=settings.CORS_ORIGINS, allow_methods=["*"], allow_headers=["*"])
app.add_middleware(TimingMiddleware)
app.add_middleware(CorrelationIdMiddleware)
# --- Exception handlers ---
@app.exception_handler(AppError)
async def app_error_handler(request: Request, exc: AppError):
return _make_error_response(exc.status_code, exc.error_code, exc.message, exc.context)
@app.exception_handler(RequestValidationError)
async def validation_error_handler(request: Request, exc: RequestValidationError):
return _make_error_response(
422,
"VALIDATION_ERROR",
"Request validation failed.",
context={"errors": exc.errors()},
)
@app.exception_handler(RateLimitExceeded)
async def rate_limit_handler(request: Request, exc: RateLimitExceeded):
return _make_error_response(429, "RATE_LIMIT_EXCEEDED", "Too many requests. Please slow down.")
@app.exception_handler(Exception)
async def unhandled_error_handler(request: Request, exc: Exception):
logger.exception("unhandled exception", extra={"path": request.url.path})
return _make_error_response(500, "INTERNAL_ERROR", "An unexpected error occurred.")
# --- Routers ---
app.include_router(health_router.router, prefix="/api/v1")
app.include_router(verification.router, prefix="/api/v1")
# --- Prometheus metrics ---
setup_metrics(app)
# --- Static UI (must be mounted last so API routes take priority) ---
import os
static_dir = os.path.join(os.path.dirname(__file__), "static")
if os.path.isdir(static_dir):
app.mount("/", StaticFiles(directory=static_dir, html=True), name="static")
return app
app = create_app()
|