FastAPI service to submit images for object detection and check task status.
- Prerequisites: Python 3.8+, Redis, Google Cloud project (Storage + Pub/Sub enabled)
- Install:
pip install -r requirements.txt- Configure env (example):
export REDIS_URL=redis://localhost:6379
export GCP_PROJECT_ID=your-gcp-project
export GCS_BUCKET=object-detection-images
export PUBSUB_TOPIC=object-detection-tasks
export TASK_TTL=3600- Run:
python -m uvicorn src.main:app --host 0.0.0.0 --port 8000Open http://localhost:8000/docs
- POST /detect: submit an image (multipart field
image)
curl -X POST http://localhost:8000/detect \
-F "image=@/path/to/image.jpg"- GET /status/{task_id}: check task status and results
curl http://localhost:8000/status/<uuid>- GET /health: lightweight health (add
?deep=1for a light GCS probe) - POST /internal/task-completed: worker callback, returns 204
- REDIS_URL: Redis connection URL
- GCP_PROJECT_ID: GCP project id
- GCS_BUCKET: GCS bucket for images
- PUBSUB_TOPIC: Pub/Sub topic for tasks
- TASK_TTL: Task TTL in seconds (default 3600)
src/domain: entities, enumssrc/application: use cases, schemassrc/infrastructure: providers, Redis/GCS/PubSubsrc/main.py: FastAPI app & routes
- Detection runs in a separate worker; this API enqueues and tracks tasks
- Blocking SDK calls are wrapped with
anyio.to_thread.run_syncto keep the event loop responsive