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
|
import anyio
|
||||||
from fastapi.responses import StreamingResponse
|
from fastapi.responses import StreamingResponse
|
||||||
|
from starlette.types import Send
|
||||||
|
|
||||||
|
|
||||||
class ChainedStream:
|
class ChainedStream:
|
||||||
@ -30,7 +31,9 @@ class ChainedStream:
|
|||||||
return await anext(self._rest)
|
return await anext(self._rest)
|
||||||
|
|
||||||
async def aclose(self) -> None:
|
async def aclose(self) -> None:
|
||||||
"""Закрыть исходный поток (идемпотентно: у завершённого генератора `aclose` — no-op)."""
|
"""Закрыть исходный поток и сбросить префетченную порцию: после закрытия `__anext__` её не отдаёт
|
||||||
|
(идемпотентно: у завершённого генератора `aclose` — no-op)."""
|
||||||
|
self._first = b""
|
||||||
await self._rest.aclose()
|
await self._rest.aclose()
|
||||||
|
|
||||||
|
|
||||||
@ -40,7 +43,7 @@ class ClosingStreamingResponse(StreamingResponse):
|
|||||||
Закрытие идёт под `CancelScope(shield=True)`: при обрыве клиента задача отдачи отменяется, а освобождение
|
Закрытие идёт под `CancelScope(shield=True)`: при обрыве клиента задача отдачи отменяется, а освобождение
|
||||||
соединения S3 требует `await`."""
|
соединения 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:
|
try:
|
||||||
await super().stream_response(send)
|
await super().stream_response(send)
|
||||||
finally:
|
finally:
|
||||||
|
|||||||
@ -178,6 +178,19 @@ async def test_chain_aclose_closes_source_when_never_iterated() -> None:
|
|||||||
assert source.closed
|
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:
|
async def test_response_closes_stream_when_client_drops_after_first_chunk() -> None:
|
||||||
"""T9.1: `send` падает на втором теле (обрыв клиента) — `body_iterator` закрыт, S3-поток освобождён."""
|
"""T9.1: `send` падает на втором теле (обрыв клиента) — `body_iterator` закрыт, S3-поток освобождён."""
|
||||||
source = _Source()
|
source = _Source()
|
||||||
|
|||||||
Loading…
x
Reference in New Issue
Block a user