Try It

Production-Hardened Version

Production-grade additions to the batch endpoint: input validation, unit tests, error tracking, and patterns for circuit breaking, caching, and rate limiting. The Interview tab is the primary submission.

UUID Input Validation IMPLEMENTED

app/routes/batch.py

# Validate order_id is a UUID before making upstream calls
import re
uuid_pattern = re.compile(
    r"^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$",
    re.IGNORECASE,
)
if not uuid_pattern.match(order_id):
    raise HTTPException(
        status_code=400,
        detail=f"Invalid order ID format. Expected UUID, got: '{order_id}'",
    )

Unit Test Suite IMPLEMENTED

test_app.py — 20 tests, all passing

# Mock strategy: patch the shared client with MockTransport
def make_mock_client(order_response, image_responses):
    transport = httpx.MockTransport(handler)
    return httpx.AsyncClient(transport=transport)

# monkeypatch replaces the module-level shared client
monkeypatch.setattr(state_module, "_http_client", make_mock_client(...))

Sentry Error Tracking IMPLEMENTED

app/__init__.py

import sentry_sdk

# Only active when DSN is configured (no-op otherwise)
if os.getenv("SENTRY_DSN"):
    sentry_sdk.init(
        dsn=os.getenv("SENTRY_DSN"),
        traces_sample_rate=0.2,
        environment=os.getenv("SENTRY_ENV", "production"),
    )

Performance Optimizations IMPLEMENTED

app/__init__.py, app/routes/batch.py

# Shared HTTP client — created once at startup
@asynccontextmanager
async def _lifespan(app: FastAPI):
    global _http_client
    _http_client = httpx.AsyncClient(timeout=60.0, follow_redirects=True)
    yield
    await _http_client.aclose()

# ZIP_STORED — no compression on already-compressed images
zipfile.ZipFile(zip_buffer, "w", zipfile.ZIP_STORED)

# Free memory incrementally as images are written
zf.writestr(filename, result["content"])
result["content"] = None

# Cap at 100 images to bound memory usage
MAX_IMAGES_PER_ORDER = 100

Runtime Stats & Monitoring IMPLEMENTED

app/state.py, app/routes/monitoring.py

# In-memory counters, incremented per request
_stats = {
    "orders_processed": 0,
    "images_downloaded": 0,
    "images_failed": 0,
    "zips_served": 0,
    "errors": [],  # last 20 errors
}

@app.get("/api/stats")
async def runtime_stats():
    return {"uptime_seconds": ..., **_stats}

Security Hardening IMPLEMENTED

app/__init__.py, app/auth.py

# CORS — restrict to known origins only
app.add_middleware(CORSMiddleware,
    allow_origins=["https://autoenhance.onrender.com", ...],
    allow_methods=["GET", "POST", "HEAD", "OPTIONS"])

# Security headers on every response
response.headers["X-Content-Type-Options"] = "nosniff"
response.headers["X-Frame-Options"] = "DENY"
response.headers["Referrer-Policy"] = "strict-origin-when-cross-origin"

# Admin token gates credit-consuming endpoints
_admin_token = os.getenv("ADMIN_TOKEN") or secrets.token_urlsafe(32)
def _require_admin(request):
    token = request.cookies.get("_at") or request.headers.get("X-Admin-Token")
    if not hmac.compare_digest(token, _admin_token): ...

# Upstream errors logged, not leaked to clients
logger.error("Upstream error: %d %s", status, text)
raise HTTPException(502, "Failed to retrieve order")

Response Caching IMPLEMENTED

app/state.py — ZIP-level cache by (order_id, format, quality, preview, dev_mode)

# state.py — TTL dict, bypassed automatically during pytest
_zip_cache: dict[tuple, dict] = {}

def get_cached_zip(key: tuple) -> dict | None:
    entry = _zip_cache.get(key)
    if entry and time.time() - entry["cached_at"] < ZIP_CACHE_TTL_SECONDS:
        return entry
    _zip_cache.pop(key, None)
    return None

# batch.py — check before downloading, store after
cache_key = (order_id, image_format, quality, preview, dev_mode)
cached = get_cached_zip(cache_key)
if cached:
    return StreamingResponse(io.BytesIO(cached["zip_bytes"]), ...)
zip_bytes, filename, headers = await _run_batch(...)
set_cached_zip(cache_key, zip_bytes, filename, headers)

Async Job Pattern for Large Orders IMPLEMENTED

app/routes/jobs.py

# 1. POST /orders/{order_id}/jobs — returns 202 immediately
@router.post("/orders/{order_id}/jobs", status_code=202)
async def create_batch_job(order_id, background_tasks, ...):
    job_id = str(uuid.uuid4())
    set_job(job_id, {"status": "processing", ...})
    background_tasks.add_task(_process_job, job_id, order_id, ...)
    return {"job_id": job_id}

# 2. GET /jobs/{job_id} — poll until "complete" or "error"
@router.get("/jobs/{job_id}")
async def get_job_status(job_id):
    job = get_job(job_id)  # returns None if expired (1h TTL)
    return {"job_id": job_id, "status": job["status"], "error": job.get("error")}

# 3. GET /jobs/{job_id}/download — fetch the ZIP
@router.get("/jobs/{job_id}/download")
async def download_job_result(job_id):
    return StreamingResponse(io.BytesIO(job["zip_bytes"]), media_type="application/zip", ...)

Circuit Breaker Pattern NEXT STEP

How it would integrate into batch.py

class CircuitBreaker:
    def __init__(self, failure_threshold=5, reset_timeout=60):
        self.failures = 0
        self.threshold = failure_threshold
        self.reset_timeout = reset_timeout
        self.state = "closed"  # closed | open | half-open
        self.last_failure_time = 0

    def record_failure(self):
        self.failures += 1
        if self.failures >= self.threshold:
            self.state = "open"
            self.last_failure_time = time.time()

    def allow_request(self) -> bool:
        if self.state == "closed": return True
        if time.time() - self.last_failure_time > self.reset_timeout:
            self.state = "half-open"
            return True
        return False

# In the endpoint:
if not circuit_breaker.allow_request():
    raise HTTPException(503, detail="Autoenhance API temporarily unavailable",
        headers={"Retry-After": "60"})

Rate Limiting Our Endpoint NEXT STEP

# Using slowapi (built on limits library)
from slowapi import Limiter
from slowapi.util import get_remote_address

limiter = Limiter(key_func=get_remote_address)

@app.get("/orders/{order_id}/images")
@limiter.limit("10/minute")
async def batch_download_order_images(request: Request, ...):
    ...