Catch asyncio.CancelledError and raise GeneratorExit in aiter functions

Aperta
  • Qualcosa non va?
#797 0 commenti 0 reazioni 0 assegnatari Vedi su GitHub

I maintainer di solito rispondono entro 1 giorno

Nessuno ha ancora preso questa issue.

Valutazione

Difficoltà
3/5
Tempo stimato
1-2 giorni
Idoneità per principianti
65/100
Tipo di issue
Bug
Chiarezza
Abbastanza chiara
Stato di attività
Tranquilla
Stack tecnologico
python
Ambito
api

Direzione di ricerca

Inizia in httpx/_client.py leggendo le funzioni aiter_raw e aiter_bytes e il modo in cui i relativi stream asincroni vengono chiusi. Riproduci l'annullamento con generator.close() o asyncio.gather(..., return_exceptions=True), quindi verifica che l'annullamento produca GeneratorExit mentre la pulizia della response avviene comunque.

Scritto dal modello di indicizzazione a partire dal testo della issue.

Descrizione

Originally opened by @gjmhmm8 on 2023-11-24 18:32:59 in encode/httpx

Hello, I am using httpx to make asynchronous requests in a generator function. However, I encountered a problem when I tried to cancel the generator using generator.close() or asyncio.gather(*tasks, return_exceptions=True). The generator did not exit gracefully and raised an asyncio.CancelledError instead of a GeneratorExit. This caused some unwanted side effects and made it difficult to handle the cancellation properly.

I looked into the source code of httpx and found that the problem was in the aiter_raw and aiter_bytes functions in the httpx/_client.py module. These functions use async for to iterate over the response stream, but they do not catch the asyncio.CancelledError that may be raised when the stream is cancelled. According to the [documentation] of asyncio.CancelledError, this exception should be caught and either re-raised or suppressed. In this case, I think it would make sense to re-raise it as a GeneratorExit, which is the expected exception for generator termination.

I suggest adding a try-except block around the async for loop in the aiter_raw and aiter_bytes functions, like this:
async def aiter_raw(
self, chunk_size: typing.Optional[int] = None
) -> typing.AsyncIterator[bytes]:
"""
A byte-iterator over the raw response content.
"""
if self.is_stream_consumed:
raise StreamConsumed()
if self.is_closed:
raise StreamClosed()
if not isinstance(self.stream, AsyncByteStream):
raise RuntimeError("Attempted to call an async iterator on an sync stream.")

    self.is_stream_consumed = True
    self._num_bytes_downloaded = 0
    chunker = ByteChunker(chunk_size=chunk_size)
    try:
        with request_context(request=self._request):
            async for raw_stream_bytes in self.stream:
                self._num_bytes_downloaded += len(raw_stream_bytes)
                for chunk in chunker.decode(raw_stream_bytes):
                    await asyncio.sleep(0)
                    yield chunk
        for chunk in chunker.flush():
            await asyncio.sleep(0)
            yield chunk
    except asyncio.CancelledError:
        raise GeneratorExit
    finally:
        await self.aclose()
Lingua principale
Python
Stelle
1.6k
Fork
103
Merge medio
1g 19h
PR unite (30g)
25

Preparare l'ambiente

Come iniziare

  1. Leggi tutta la issue e poi la guida ai contributi del progetto.
  2. Commenta sulla issue per dire che te ne occupi tu — evita che due persone facciano lo stesso lavoro.
  3. Fai un fork del repository e lavora su un branch.
  4. Apri una pull request che faccia riferimento al numero della issue.

Altre issue di pydantic/httpx2

Tutte le issue di pydantic/httpx2

Issue simili

Altre issue su Python

Ricevi le nuove issue nella tua casella

Un breve riepilogo di issue GitHub adatte ai principianti.