generated from john/python-template
Step 3 tested with real data. Added Step 4 implementation plan
This commit is contained in:
Jim Lancaster
+174
-102
@@ -1,18 +1,18 @@
|
||||
## Step 4: `services/upload.py` + `worker.py`
|
||||
|
||||
## Objective
|
||||
### Objective
|
||||
|
||||
Implement the MVP upload and background processing pipeline so the system can:
|
||||
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
|
||||
- call Step 3 transcription service
|
||||
- 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.
|
||||
This step advances MVP Feature 1 + Feature 2 and supports REQ-1, REQ-2, REQ-3, REQ-4, REQ-6.
|
||||
|
||||
---
|
||||
|
||||
@@ -21,15 +21,15 @@ This covers MVP Feature 1 + 2 and supports REQ-1, REQ-2, REQ-3, REQ-4, REQ-6.
|
||||
### 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)
|
||||
- Upload persistence logic and initial job creation
|
||||
- Worker polling and single-job lifecycle execution
|
||||
- Deterministic test coverage for upload + worker (default suite)
|
||||
|
||||
### Out of scope
|
||||
- UI wiring (Step 5)
|
||||
- Job/transcript pages (Step 5)
|
||||
- External queue infrastructure
|
||||
- Async DB/session architecture changes (post-MVP)
|
||||
- UI integration and pages (Step 5)
|
||||
- Queue infrastructure beyond in-process loop
|
||||
- Async DB/session architecture refactor
|
||||
- Broad production hardening beyond MVP needs
|
||||
|
||||
---
|
||||
|
||||
@@ -40,151 +40,223 @@ This covers MVP Feature 1 + 2 and supports REQ-1, REQ-2, REQ-3, REQ-4, REQ-6.
|
||||
- `src/transcription/worker.py`
|
||||
- `src/transcription/services/__init__.py` (export updates as needed)
|
||||
|
||||
### Tests
|
||||
### Test files
|
||||
- `tests/services/test_upload.py`
|
||||
- `tests/services/test_worker.py` *(recommended for mirror consistency)*
|
||||
- `tests/services/test_worker.py`
|
||||
|
||||
### Optional external lane (already present pattern)
|
||||
- reuse `external` marker for live-provider checks where appropriate
|
||||
- keep external out of default lane
|
||||
|
||||
---
|
||||
|
||||
## Required MCP Prompt References (for test workflow)
|
||||
|
||||
Apply these resources directly during Step 4 test creation:
|
||||
|
||||
1. `resource://catalog/prompts/pytest-scaffold`
|
||||
2. `resource://prompts/pytest-scaffold/document`
|
||||
3. `resource://catalog/prompts/pytest-fill-scaffold`
|
||||
4. `resource://prompts/pytest-fill-scaffold/document`
|
||||
|
||||
And (as referenced by those prompts) apply relevant pytest skill references for:
|
||||
- naming/hierarchy
|
||||
- marker defaults
|
||||
- SQLAlchemy sync testing behavior where applicable
|
||||
|
||||
---
|
||||
|
||||
## Design Decisions
|
||||
|
||||
1. **Upload service owns file persistence + initial DB records**
|
||||
- Save file first, then create `Document` + `Job(queued)`.
|
||||
1. **Upload service owns initial file + record creation**
|
||||
- Writes file, creates `Document`, creates queued `Job`, returns IDs/path.
|
||||
|
||||
2. **Worker owns job lifecycle transitions**
|
||||
- Transition only in worker:
|
||||
- `queued -> processing`
|
||||
- `processing -> transcribed|failed`
|
||||
2. **Worker owns lifecycle transitions**
|
||||
- Worker is the single owner of `queued -> processing -> terminal` job state changes.
|
||||
|
||||
3. **Worker calls Step 3 service, not provider SDK directly**
|
||||
- Use `transcribe_document_image(...)` from `services/transcription.py`.
|
||||
3. **Worker uses Step 3 service boundary**
|
||||
- Worker calls `transcribe_document_image(...)`; no provider-specific SDK logic in worker.
|
||||
|
||||
4. **Failure details are always persisted**
|
||||
- On failure, persist transcript row with `text=None`, `error_detail=...`.
|
||||
4. **Failure information is always persisted**
|
||||
- On failure: store `Transcript(text=None, error_detail=...)` and set `Job.status=failed`.
|
||||
|
||||
5. **Worker loop is stoppable**
|
||||
- Simple in-process loop with stop signal/event and poll interval.
|
||||
5. **Loop remains simple and stoppable**
|
||||
- In-process polling loop with stop event and poll interval for MVP simplicity and testability.
|
||||
|
||||
---
|
||||
|
||||
## Task-by-Task Execution Checklist
|
||||
|
||||
## Phase A — Upload service (`services/upload.py`)
|
||||
## Phase A — Implement upload service (`src/transcription/services/upload.py`)
|
||||
|
||||
- [ ] Create `src/transcription/services/upload.py`
|
||||
- [ ] Add `UploadError` exception type
|
||||
- [ ] Add result dataclass (e.g., `UploadJobResult`) with:
|
||||
- [ ] Create `UploadError` exception
|
||||
- [ ] Create `UploadJobResult` dataclass 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)
|
||||
- [ ] Add filename safety handling:
|
||||
- [ ] normalize to basename
|
||||
- [ ] avoid path traversal
|
||||
- [ ] collision-safe stored name (e.g., UUID prefix/suffix)
|
||||
- [ ] Validate upload payload:
|
||||
- [ ] non-empty bytes required
|
||||
- [ ] extension in supported set (`.jpg/.jpeg/.png/.tif/.tiff/.pdf`)
|
||||
- [ ] Ensure upload directory exists (`mkdir(parents=True, exist_ok=True)`)
|
||||
- [ ] Write file bytes to `UPLOAD_DIR`
|
||||
- [ ] Persist DB records in one transaction:
|
||||
- [ ] `Document(filename, file_path)`
|
||||
- [ ] `Job(document_id=..., status=queued)`
|
||||
- [ ] Return `UploadJobResult`
|
||||
- [ ] Add logging for success/failure boundaries
|
||||
|
||||
---
|
||||
|
||||
## Phase B — Worker core (`worker.py`)
|
||||
## Phase B — Implement worker core (`src/transcription/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
|
||||
- [ ] Add `process_next_queued_job(...) -> bool`
|
||||
- [ ] Fetch oldest queued job
|
||||
- [ ] Return `False` when no queued jobs exist
|
||||
- [ ] Transition picked job to `processing` and update timestamp
|
||||
- [ ] Resolve associated `Document.file_path`
|
||||
- [ ] Call `transcribe_document_image(image_path=...)`
|
||||
- [ ] On success:
|
||||
- [ ] insert/update transcript text
|
||||
- [ ] clear error detail
|
||||
- [ ] mark job `transcribed`
|
||||
- [ ] update timestamp
|
||||
- [ ] On failure:
|
||||
- [ ] insert/update transcript with `text=None`, `error_detail=...`
|
||||
- [ ] mark job `failed`
|
||||
- [ ] update timestamp
|
||||
- [ ] Commit terminal state and return `True`
|
||||
- [ ] Add logs around job pickup, transition, and terminal outcome
|
||||
|
||||
---
|
||||
|
||||
## Phase C — Worker loop (`worker.py`)
|
||||
## Phase C — Implement worker loop (`src/transcription/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
|
||||
- [ ] Add `run_worker_loop(...)`
|
||||
- [ ] Accept configurable stop event/signal
|
||||
- [ ] Accept configurable poll interval
|
||||
- [ ] Repeatedly call `process_next_queued_job`
|
||||
- [ ] Sleep only when queue is empty
|
||||
- [ ] Exit cleanly when stop event is set
|
||||
|
||||
---
|
||||
|
||||
## Phase D — Service exports
|
||||
## Phase D — Exports
|
||||
|
||||
- [ ] Update `src/transcription/services/__init__.py` exports
|
||||
- [ ] include upload service symbols
|
||||
- [ ] keep transcription exports intact
|
||||
- [ ] Update `src/transcription/services/__init__.py` to expose upload APIs
|
||||
- [ ] Keep existing transcription exports intact
|
||||
|
||||
---
|
||||
|
||||
## Phase E — Tests (scaffold -> fill)
|
||||
## Phase E — Tests via MCP scaffold -> fill flow
|
||||
|
||||
### 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`
|
||||
## E1 Scaffold (structure only)
|
||||
|
||||
### E2 Fill
|
||||
Use scaffold prompt workflow first for:
|
||||
- `src/transcription/services/upload.py`
|
||||
- `src/transcription/worker.py`
|
||||
|
||||
#### 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`)
|
||||
Expected scaffold targets:
|
||||
- `tests/services/test_upload.py`
|
||||
- `tests/services/test_worker.py`
|
||||
|
||||
#### 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`)
|
||||
Scaffold rules:
|
||||
- [ ] Class hierarchy + method names + one-line docstrings only
|
||||
- [ ] No assertions or implementation details in scaffold phase
|
||||
- [ ] Keep method names concise and behavior-focused
|
||||
|
||||
### 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)
|
||||
Validation:
|
||||
- [ ] `uv run pytest --collect-only -q`
|
||||
|
||||
## E2 Fill scaffold (implementation)
|
||||
|
||||
Use fill prompt workflow for:
|
||||
- `tests/services/test_upload.py`
|
||||
- `tests/services/test_worker.py`
|
||||
- stack: `sqlalchemy-sync` (or `mixed` if combining pure + DB behaviors)
|
||||
- marker lane preference: `unit` and `integration` as appropriate
|
||||
- strategy: minimal deterministic implementation
|
||||
|
||||
Fill rules (invariants):
|
||||
- [ ] Preserve scaffold class names, method names, and one-line docstrings
|
||||
- [ ] Do not rename/re-nest scaffolded tests unless explicitly approved
|
||||
- [ ] One behavior target per test
|
||||
- [ ] Minimal mocking; mock only network/nondeterministic boundaries
|
||||
|
||||
Suggested test coverage:
|
||||
|
||||
### `tests/services/test_upload.py`
|
||||
- [ ] creates file + document + queued job (`integration`)
|
||||
- [ ] rejects empty bytes (`unit`)
|
||||
- [ ] rejects unsupported extension (`unit`)
|
||||
- [ ] writes collision-safe unique filename (`integration`)
|
||||
- [ ] persisted job status is `queued` (`integration`)
|
||||
|
||||
### `tests/services/test_worker.py`
|
||||
- [ ] returns `False` when queue empty (`integration`)
|
||||
- [ ] transitions `queued -> processing -> transcribed` on success (`integration`)
|
||||
- [ ] stores transcript text on success (`integration`)
|
||||
- [ ] transitions to `failed` and stores `error_detail` on failure (`integration`)
|
||||
- [ ] updates existing transcript instead of duplicate create (`integration`)
|
||||
- [ ] worker loop exits when stop event set (`unit`)
|
||||
|
||||
---
|
||||
|
||||
## Phase F — Verification
|
||||
## Marker Strategy
|
||||
|
||||
- `unit`: pure logic tests (filename handling, loop stop behavior, validation logic)
|
||||
- `integration`: DB + service orchestration tests (SQLite/session/contracts)
|
||||
- `external`: opt-in live provider tests only (not part of default Step 4 lane)
|
||||
|
||||
No new marker needed; reuse existing marker registration.
|
||||
|
||||
---
|
||||
|
||||
## Validation Sequence (strict order)
|
||||
|
||||
- [ ] `uv run pytest --collect-only -q`
|
||||
- [ ] `uv run pytest -m unit -q` *(if unit tests touched)*
|
||||
- [ ] `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
|
||||
## Reporting Requirements (after implementation)
|
||||
|
||||
- [ ] 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)
|
||||
Implementation report must include:
|
||||
|
||||
1. Files created/updated
|
||||
2. Fixture and marker decisions
|
||||
3. MCP references used and why
|
||||
4. Validation command results
|
||||
5. Remaining risks/open questions (only blockers)
|
||||
|
||||
---
|
||||
|
||||
## Guardrails
|
||||
|
||||
- Keep Step 4 independent from UI concerns.
|
||||
- Do not call provider SDK directly from worker.
|
||||
- Do not silently swallow exceptions.
|
||||
- Always persist terminal job outcome.
|
||||
- Keep default suite deterministic and fast.
|
||||
- Preserve scaffold invariants during fill phase.
|
||||
|
||||
---
|
||||
|
||||
## 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`
|
||||
Step 4 is complete when:
|
||||
|
||||
- [ ] Upload service writes file and creates `Document` + queued `Job`
|
||||
- [ ] Worker processes queued jobs end-to-end using Step 3 transcription service
|
||||
- [ ] Success path persists transcript text and sets `transcribed`
|
||||
- [ ] Failure path persists error detail and sets `failed`
|
||||
- [ ] Queue-empty path returns cleanly
|
||||
- [ ] New tests pass and full suite is green (`uv run pytest -q`)
|
||||
- [ ] Output report includes MCP reference usage + validation evidence
|
||||
Reference in New Issue
Block a user