transcription/docs/step4.md

Step 4: services/upload.py + worker.py

Objective

Implement the MVP upload and background processing pipeline so the system can:

1. Save uploaded files into UPLOAD_DIR
2. Create Document + Job(status="queued")
3. Process queued jobs in a worker loop:
   • queued -> processing
   • invoke Step 3 transcription
   • persist Transcript
   • finalize as transcribed or failed

This covers MVP Feature 1 + 2 and supports REQ-1, REQ-2, REQ-3, REQ-4, REQ-6.

---

Scope

In scope

• src/transcription/services/upload.py
• src/transcription/worker.py
• Upload persistence logic + job creation
• Worker polling and single-job processing lifecycle
• Tests for upload + worker behavior (deterministic, no live API)

Out of scope

• UI wiring (Step 5)
• Job/transcript pages (Step 5)
• External queue infrastructure
• Async DB/session architecture changes (post-MVP)

---

Planned Deliverables

Source files

• src/transcription/services/upload.py
• src/transcription/worker.py
• src/transcription/services/__init__.py (export updates as needed)

Tests

• tests/services/test_upload.py
• tests/services/test_worker.py (recommended for mirror consistency)

---

Design Decisions

1. Upload service owns file persistence + initial DB records

   • Save file first, then create Document + Job(queued).

2. Worker owns job lifecycle transitions

   • Transition only in worker:
     • queued -> processing
     • processing -> transcribed|failed

3. Worker calls Step 3 service, not provider SDK directly

   • Use transcribe_document_image(...) from services/transcription.py.

4. Failure details are always persisted

   • On failure, persist transcript row with text=None, error_detail=....

5. Worker loop is stoppable

   • Simple in-process loop with stop signal/event and poll interval.

---

Task-by-Task Execution Checklist

Phase A — Upload service (services/upload.py)

• Create src/transcription/services/upload.py
• Add UploadError exception type
• Add result dataclass (e.g., UploadJobResult) with:
  • document_id
  • job_id
  • stored_path
  • original_filename
• Implement safe filename handling:
  • basename normalization
  • collision-safe naming (e.g., UUID prefix)
  • reject path traversal patterns
• Implement extension validation (.jpg/.jpeg/.png/.tif/.tiff/.pdf)
• Implement empty-bytes validation
• Ensure UPLOAD_DIR exists (mkdir(parents=True, exist_ok=True))
• Persist file bytes to disk
• Insert Document and Job(status=queued) in one transaction
• Return structured UploadJobResult
• Add logging for upload success/failure (no sensitive payload logging)

---

Phase B — Worker core (worker.py)

• Create src/transcription/worker.py
• Implement process_next_queued_job(...) -> bool
  • fetch oldest queued job
  • return False if none found
  • set job to processing + update timestamp
  • call transcribe_document_image(image_path=...)
  • success path: persist transcript text, mark transcribed
  • failure path: persist error detail, mark failed
  • return True when a job is processed
• Implement transcript upsert behavior (avoid duplicate unique job_id insert issues)
• Add structured logs for lifecycle transitions and outcomes

---

Phase C — Worker loop (worker.py)

• Implement run_worker_loop(...)
  • accepts stop signal/event
  • configurable poll_interval_seconds
  • repeatedly calls process_next_queued_job
  • sleeps when no work
  • exits cleanly on stop request
• Keep implementation single-process/simple for MVP assumptions

---

Phase D — Service exports

• Update src/transcription/services/__init__.py exports
  • include upload service symbols
  • keep transcription exports intact

---

Phase E — Tests (scaffold -> fill)

E1 Scaffold

• Create tests/services/test_upload.py with class/method skeletons + one-line docstrings
• Create tests/services/test_worker.py with class/method skeletons + one-line docstrings
• Validate scaffold collection:
  • uv run pytest --collect-only -q

E2 Fill

Upload tests

• test_create_upload_job_writes_file_and_creates_records (integration)
• test_create_upload_job_rejects_empty_bytes (unit)
• test_create_upload_job_rejects_unsupported_extension (unit)
• test_create_upload_job_uses_unique_stored_filename (integration)
• test_create_upload_job_sets_job_status_queued (integration)

Worker tests

• test_process_next_queued_job_returns_false_when_queue_empty (integration)
• test_process_next_queued_job_transitions_processing_to_transcribed (integration)
• test_process_next_queued_job_persists_transcript_text_on_success (integration)
• test_process_next_queued_job_sets_failed_and_error_detail_on_failure (integration)
• test_process_next_queued_job_updates_existing_transcript_if_present (integration)
• test_run_worker_loop_stops_when_stop_event_is_set (unit)

Marker usage

• Mark pure logic tests as @pytest.mark.unit
• Mark DB/service orchestration tests as @pytest.mark.integration
• Reuse existing external only for future live endpoint checks (not required in Step 4)

---

Phase F — Verification

• uv run pytest --collect-only -q
• uv run pytest tests/services/test_upload.py -q
• uv run pytest tests/services/test_worker.py -q
• uv run pytest -m "unit or integration" -q
• uv run pytest -q

---

Implementation Guardrails

• Do not couple worker directly to provider SDK internals
• Do not swallow exceptions silently
• Always persist terminal failure details
• Keep status transitions explicit and timestamped
• Keep default test suite deterministic and fast (no live network)

---

Definition of Done (Step 4)

• Upload service persists file + creates Document and queued Job
• Worker processes queued jobs end-to-end via Step 3 service
• Success writes transcript text and final transcribed status
• Failure writes error_detail and final failed status
• Queue-empty worker call returns cleanly without error
• Full test suite passes with uv run pytest -q