adminandClaude
ec6fd7157a
Surveyed ~/dev (mewtwo) + jirachi. Battery mirrors actual workload:
mcp_server -> 9 MCP servers (joplin/obsidian/vault/project-rag/...)
tts_pipeline -> TTS/audio pipelines (Chatterbox, aygea-tts, vr-to-tts)
webhook_bridge -> Twitch/Discord bridges (multistream, notifier, overlay)
data_service -> data/API (Supabase MCP, PostgresHA, dashboard)
automation_glue -> batch/cron glue (fix-tokens, notesCleanup)
rust_service -> big Rust services (NineSentry, aystreamer): tokio
channels + Arc/Mutex + error enums + graceful shutdown
Each prompt is ~2-3KB (fits 128k context with output room), single-file,
runnable, graded on the same 5-pillar rubric. aygea_test_battery.md is the
index + scoring notes + the prompt_id schema the dashboard will need for
multi-prompt support.
Co-Authored-By: Claude <noreply@anthropic.com>
31 lines
2.6 KiB
Plaintext
31 lines
2.6 KiB
Plaintext
Write a complete, single-file data-access service (Python 3.11+ asyncio, stdlib only) that wraps a **mock** Postgres connection pool. It must demonstrate correct pooling, parameterized queries, pagination, and transactional integrity.
|
|
|
|
### Mock layer
|
|
Provide an in-memory mock that stands in for a real pool. Something like:
|
|
`async def acquire() -> Connection` / `connection.release()`, where `Connection` has
|
|
`async def fetch(query, params)` and `async def execute(query, params)`. Seed it with ~25
|
|
"users" and ~5 "posts" per user. Make `acquire()` sometimes wait briefly and make it
|
|
**track how many connections are checked out** so leaks are detectable.
|
|
|
|
### Service API
|
|
1. `async def get_users(page: int, page_size: int) -> dict` — returns `{"items": [...], "page": p, "page_size": s, "total": N}`. Validate page/page_size (page ≥ 1, 1 ≤ page_size ≤ 100).
|
|
2. `async def get_user_with_posts(user_id: int) -> dict` — one user + their posts (efficient: don't N+1).
|
|
3. `async def relabel_users(pairs: list[tuple[int, str]]) -> int` — bulk update each user's `label`. Transactional: if ANY update fails (e.g. an id doesn't exist), the whole batch rolls back and returns 0 (or raises) — never partially applied.
|
|
|
|
### Requirements
|
|
1. **Connection-pool discipline:** every acquire is paired with a release in a `finally`/context manager. The service must prove across 100 calls that connections-in-use returns to 0 (no leak).
|
|
2. **Parameterized queries:** all queries pass values as parameters (mock `(query, params)`), NEVER string-interpolated. Include at least one test that would "fail" on injection if interpolation were used.
|
|
3. **Pagination correctness:** `total` is the true count regardless of page; last page returns the correct remainder; out-of-range page returns empty (not an error).
|
|
4. **Transactional bulk:** `relabel_users` wraps all updates in one transaction; a partial failure rolls back. Prove it.
|
|
5. **Acquire timeout:** `acquire()` must respect a max-wait (e.g. 2s) and raise a clear error if the pool is exhausted, rather than hanging.
|
|
6. **No silent failures:** missing user → clear `KeyError`-style result; bad params → validation error.
|
|
|
|
### Included tests
|
|
Include a runnable test section that asserts:
|
|
- a) 100 sequential `get_users` calls leave the pool at 0 checked-out (no leak).
|
|
- b) Pagination math: total matches seed count; last page has the right remainder.
|
|
- c) `relabel_users` with one bad id rolls back — verify NO labels changed afterward.
|
|
- d) Concurrent `get_user_with_posts` for 10 users at once completes without pool exhaustion.
|
|
|
|
Provide clean, well-commented code that runs directly via `python file.py`.
|