Skip to content

Add pgdevkit[fastapi]: streaming Postgres JSON response with query cancellation - #48

Merged
aersam merged 7 commits into
mainfrom
fastapi-json-response
Oct 9, 2026
Merged

aersam merged 7 commits into
mainfrom
fastapi-json-response

Conversation

@aersam

@aersam aersam commented Oct 9, 2026 •

Copy link
Copy Markdown
Contributor

Summary

Adds the optional pgdevkit[fastapi] extra with pgdevkit.fastapi.PostgresJsonResponse: a port of CCMT2's PostgresJsonResponse that CCMT2, OneSales and MDMApp can share. It streams a query as a JSON array (Postgres builds the JSON via row_to_json; rows are streamed with cursor.stream) and cancels the query in Postgres when the client disconnects.

set_default_pool(pool)  # once at startup, shared with fetch_all

@router.get("/articles", responses={200: {"model": list[ArticleOut]}})
async def articles(lng: str) -> PostgresJsonResponse:
    return PostgresJsonResponse("select ... where lng = %(lng)s", {"lng": lng})
  • Same connection handling as fetch_all (con= / pool= / set_default_pool()), reusing its _acquire, _render and types.
  • Cancellation: own http.disconnect watcher (Starlette only listens on ASGI spec < 2.4), then a shielded cancel_safe() + close of a connection left mid-query. Only a still-running query is cancelled, so a late cancel can't hit the next statement on a pooled connection.
  • Errors before the first byte -> JSON 500 (text hidden unless a subclass sets expose_errors); after the first byte the array stays unterminated so clients can't mistake it for complete.
  • Optional statement_timeout=<seconds> on both PostgresJsonResponse and fetch_all: Postgres aborts the query (SET LOCAL semantics: transaction-scoped, never leaks into the pool, previous value restored on a borrowed con). fetch_all raises QueryCanceled; the response sends a 504 before the first byte. It bounds query execution only: a client that stopped reading keeps its backend (ClientWrite) regardless, so a proxy write timeout is still needed.
  • Version 0.14.0 (main is at 0.13.0), fastapi >= 0.118 (a Depends with yield only releases its connection after the streaming response from 0.118).

Findings worth knowing (all covered by tests)

  • Without an explicit cancel, Postgres keeps running the query after the pool discards the connection (pg_sleep(30) stays active); psycopg only cancels on native asyncio cancellation, not when an anyio scope re-cancels its cleanup.
  • An anyio task group around the body (also stock Starlette StreamingResponse) makes Granian hang when the body raises mid-stream, so the disconnect watcher uses an asyncio.timeout scope instead.
  • Postgres flushes rows in 8 kB buffers, so a slow query with tiny rows delivers late.

Test plan

  • tests/test_pg_json_response.py: output shape (empty / exact batch multiples / multi-batch), params, composables, query_produces_json, headers, default pool, con/pool, dict_row connection, validation, errors before/after the first byte
  • tests/test_pg_json_cancel.py: disconnect before first byte / while streaming, task cancellation, failing send, anyio scope cancellation, a second cancel during cleanup; each verified in pg_stat_activity
  • tests/test_pg_json_granian.py: real Granian server, aborted HTTP client, mid-stream error
  • Mutation-checked: disabling the shielded cancel fails the scope-cancellation tests
  • tests/db/test_fetch_statement_timeout.py, tests/test_pg_json_statement_timeout.py: timeout before/after the first byte, no leak into the pool, previous setting restored, autocommit, validation (mutation-checked)
  • tests/db/test_fetch.py (from Add fetch_all with optional borrowed connection and Pydantic model overload #47) needs Docker and could not run in my sandbox: CI

Not in this PR

Migrating CCMT2 / OneSales (backend/db/pg_response.py, notifications.py) / MDMApp call sites, after release. A write timeout for slow readers is documented (proxy + pool sizing), not implemented: statement_timeout doesn't cover it.

🤖 Generated with Claude Code

@aersam

aersam commented Oct 9, 2026

Copy link
Copy Markdown
Contributor Author

Implementation plan (for review before coding)

Attachments

Claude Session: c9be081f-307e-54df-85ad-8cc47a4eda79

@aersam

aersam commented Oct 9, 2026

Copy link
Copy Markdown
Contributor Author

Plan review amendments (bms-plan-review: go with changes)

Attachments

Claude Session: c9be081f-307e-54df-85ad-8cc47a4eda79

@aersam
aersam marked this pull request as ready for review October 9, 2026 13:14
@aersam
aersam merged commit 3dd0479 into main Oct 9, 2026
2 checks passed
aersam and others added 7 commits October 9, 2026 14:50
Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
… on client disconnect

Optional extra pgdevkit[fastapi]. Port of CCMT2's PostgresJsonResponse, with a
connection-source API usable from CCMT2, OneSales and MDMApp, true row streaming
(cursor.stream), own http.disconnect detection (independent of ASGI spec version)
and shielded cancel_safe() so Postgres stops the query. Bump to 0.13.0.

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
… connection handling

PostgresJsonResponse now takes (query, params, *, con=, pool=) like fetch_all, reuses its
connection acquisition (set_default_pool), query rendering (sqlglot support) and types.
Version 0.14.0 (main is at 0.13.0).

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
…nterruptible cancel cleanup, FastAPI >= 0.118, docs

- cursor uses tuple_row so a dict_row connection/pool can't corrupt the output
- batch_size >= 1 and t-strings are rejected at construction
- cancel cleanup runs as its own shielded task: a second (native) cancellation can no longer skip cancel+close
- only send() failures count as a client disconnect; HTTPException headers kept
- fastapi >= 0.118 (Depends with yield releases after the streaming response); extra = pgdevkit[db] + fastapi
- docs: con= semantics, OpenAPI responses=, first-byte timing, expose_errors warning, slow-reader note

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
Postgres enforces it (set_config(..., is_local => true)): transaction-scoped, restored on a
borrowed connection, never leaks into a pool. fetch_all raises QueryCanceled; the response
answers 504 before the first byte. It does not free a stalled reader (documented, verified).

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

1 participant