Skip to content

🔄 Fetch Request Workflow

A fetch request is a statement ingestion job. This is the end-to-end journey and how to drive it.

📥 Creating a Fetch Request

Via API:

1
2
3
4
# File source
curl -X POST http://localhost:8000/fetch-requests \
  -H "Content-Type: application/json" \
  -d '{"user_id":"...","account_id":"...","source":{"type":"file","path":"/var/statements/axis.pdf","bank":"axis"},"start_date":"2025-04-01","end_date":"2025-04-30"}'

Via CLI:

1
2
3
4
# from repo root, with the khata venv active
.venv\Scripts\python.exe -m interfaces.fetch_request.cli create-file --account "Axis Savings" --path /var/statements/axis.pdf --bank axis
.venv\Scripts\python.exe -m interfaces.fetch_request.cli list --status pending
.venv\Scripts\python.exe -m interfaces.fetch_request.cli get <id>

CLI subcommands: list, get, create-file, create-email, ambiguities, resolve, retry.

⚙️ The Worker

worker.py polls for pending fetch requests and runs the pipeline.

1
.venv\Scripts\python.exe worker.py

Knobs (env vars):

Env var Default Meaning
WORKER_POLL_MIN 5 Min poll interval (seconds); backs off ×2 up to MAX when idle
WORKER_POLL_MAX 60 Max poll interval
WORKER_MAX_RETRIES 3 Failed requests stop being retried beyond this count

Per cycle: fetch status="pending" (limit 100), mark each processing, run the orchestrator, then:

  • success → completed,
  • PauseForResolution → paused (retry count reset),
  • exception → failed with retry_count + 1.

Ambiguity resolution is enabled only when an LLM_API_KEY is present; today worker._maybe_create_resolver() returns None.

📡 Progress & SSE

ProgressEventBus (interfaces/sse/) emits typed progress events during a run; clients subscribe to the SSE endpoint per fetch request to watch stage transitions. The domain declares the event model in domain/fetch_pipeline/messaging/ (see ProgressEventBus port).

🧠 Orchestration

PipelineOrchestrator (domain/fetch_pipeline/utils/pipeline.py) wires the worker's services:

1
2
fr_svc · ambiguity_svc · expense_svc · entity_svc · tag_svc
pdf/email/xlsx/csv fetchers · sse bus · http cache store · resolved_choices

with pause_on_ambiguity=True and a pre-loaded map of previously resolved ambiguity choices so repeats don't re-pause.

Extraction Pipelines · Domain Resources · Running The Service