fix(files): типизация send и сброс префетченной порции в ChainedStream.aclose (PB.1, PB.2)
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_012vJkmBGKKqMoEMzvsXG3y2
This commit is contained in:
parent
7578b5fb12
commit
308fbc4fb4
@ -10,6 +10,7 @@ from collections.abc import AsyncGenerator
|
||||
|
||||
import anyio
|
||||
from fastapi.responses import StreamingResponse
|
||||
from starlette.types import Send
|
||||
|
||||
|
||||
class ChainedStream:
|
||||
@ -30,7 +31,9 @@ class ChainedStream:
|
||||
return await anext(self._rest)
|
||||
|
||||
async def aclose(self) -> None:
|
||||
"""Закрыть исходный поток (идемпотентно: у завершённого генератора `aclose` — no-op)."""
|
||||
"""Закрыть исходный поток и сбросить префетченную порцию: после закрытия `__anext__` её не отдаёт
|
||||
(идемпотентно: у завершённого генератора `aclose` — no-op)."""
|
||||
self._first = b""
|
||||
await self._rest.aclose()
|
||||
|
||||
|
||||
@ -40,7 +43,7 @@ class ClosingStreamingResponse(StreamingResponse):
|
||||
Закрытие идёт под `CancelScope(shield=True)`: при обрыве клиента задача отдачи отменяется, а освобождение
|
||||
соединения S3 требует `await`."""
|
||||
|
||||
async def stream_response(self, send) -> None: # type: ignore[no-untyped-def] # noqa: ANN001
|
||||
async def stream_response(self, send: Send) -> None:
|
||||
try:
|
||||
await super().stream_response(send)
|
||||
finally:
|
||||
|
||||
@ -178,6 +178,19 @@ async def test_chain_aclose_closes_source_when_never_iterated() -> None:
|
||||
assert source.closed
|
||||
|
||||
|
||||
async def test_chain_does_not_yield_prefetched_chunk_after_aclose() -> None:
|
||||
"""PB.2: закрытая обёртка не отдаёт префетченную порцию (иначе потребитель читал бы данные закрытого потока)."""
|
||||
source = _Source()
|
||||
first = await anext(source.iterator, b"")
|
||||
assert first == b"x"
|
||||
chain = _chain(first, source.iterator)
|
||||
await chain.aclose()
|
||||
assert source.closed
|
||||
with pytest.raises(StopAsyncIteration):
|
||||
await chain.__anext__()
|
||||
await chain.aclose() # повторное закрытие безопасно
|
||||
|
||||
|
||||
async def test_response_closes_stream_when_client_drops_after_first_chunk() -> None:
|
||||
"""T9.1: `send` падает на втором теле (обрыв клиента) — `body_iterator` закрыт, S3-поток освобождён."""
|
||||
source = _Source()
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user