Stream
Local development
Run, debug, and test Stream locally
Prerequisites
uvand Python 3.13.- GCP credentials with access to Spanner, Cloud Tasks, and Secret Manager.
- Pub/Sub push requests must be simulated or replayed for local testing.
Required environment
export GCP_PROJECT_ID="..."
export SPANNER_INSTANCE_ID="..."
export SPANNER_DATABASE_ID="..."
export CLOUD_TASKS_SERVICE_ACCOUNT="..."
export CLOUD_TASKS_LOCATION="..."
export EZYVET_ETL_QUEUE="..."
export BITWERX_ETL_QUEUE="..."
export HAP_ETL_QUEUE="..."
export HANDLER_SERVICE_BASE_URL="https://<handler-cloud-run-url>"
export GOOGLE_APPLICATION_CREDENTIALS="/path/to/service-account.json"Install dependencies
# From repo root
uv sync --all-packagesRun locally
# From repo root
uv run --package stream granian --interface asgi services.stream.main:app --host 0.0.0.0 --port 8080 --workers 1 --loop uvloopSanity checks
curl http://localhost:8080/api/v1/healthExample request
curl \
-H "Content-Type: application/json" \
-d '{"commitTimestamp": {"seconds": 1, "nanos": 0}, "mods": []}' \
http://localhost:8080/api/v1/pipeline/streamTests
# From repo root
uv run --package stream pytest -qLast updated on