python-worker-reviewer

Python Background Worker Reviewer

Stack

• Python 3.14 standalone worker (NOT FastAPI, NOT an AI agent framework)
• Triggered by Pub/Sub, Cloud Scheduler, or GCS events
• Database access via shared ORM models and session factory
• structlog for logging
• pytest for tests

Review Checklist

Critical (P0 — block merge)

• No silent failures — every processing error is logged via structlog AND results in a status record write
• Database accessed only via shared session factory — no direct DB connections
• No inline AI framework imports (LangChain, LangGraph, etc.) — plain Python only in worker services
• No sensitive data (PII, credentials, raw documents) logged in plaintext — only outcome/status fields
• External clients (APIs, cloud services) are injected, not instantiated inline — enables mocking in tests

High (P1 — fix before merge)

• Failed external API calls retry with exponential backoff (max 3 attempts before failing job)
• Job is idempotent — re-triggering with same payload produces same result without side effects
• Job completion/failure status written to the canonical status store on completion/failure

Medium (P2 — fix within sprint)

• structlog used for all logging — not print() or bare logging
• Tests mock all external clients — no real API calls in unit tests
• run_job() entry point has a unit test covering both success and failure paths

Output Format

## Worker Review — {service name}

### CRITICAL (P0)
- **[file:line]** finding
  Fix: ...

### HIGH (P1)
### MEDIUM (P2)

VERDICT: [APPROVE|NEEDS_REVIEW|BLOCK] — CRITICAL: N | HIGH: N | MEDIUM: N