Catch asyncio.CancelledError and raise GeneratorExit in aiter functions

Abierto
  • ¿Algo no está bien?
#797 0 comentarios 0 reacciones 0 asignados Ver en GitHub

Los mantenedores suelen responder en 1 día

Nadie ha tomado este issue todavía.

Evaluación

Dificultad
3/5
Tiempo estimado
1-2 días
Aptitud para principiantes
65/100
Tipo de issue
Error
Claridad
Bastante claro
Estado de actividad
Tranquilo
Stack tecnológico
python
Área
api

Línea de trabajo

Comienza en httpx/_client.py leyendo las funciones aiter_raw y aiter_bytes y cómo se cierran sus streams asíncronos. Reproduce la cancelación con generator.close() o asyncio.gather(..., return_exceptions=True) y, después, verifica que la cancelación produzca GeneratorExit mientras la limpieza de la response siga realizándose.

Escrito por el modelo de indexación a partir del texto del issue.

Descripción

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()
Lenguaje dominante
Python
Estrellas
1.6k
Forks
103
Merge medio
1 d 19 h
PR fusionados (30 d)
25

Preparar el entorno

Primeros pasos

  1. Lee el issue completo y luego la guía de contribución del proyecto.
  2. Comenta en el issue que vas a ocuparte — evita que dos personas hagan lo mismo.
  3. Haz un fork del repositorio y trabaja en una rama.
  4. Abre un pull request que haga referencia al número del issue.

Más de pydantic/httpx2

Todos los issues de pydantic/httpx2

Issues similares

Más issues de Python

Recibe los nuevos issues en tu correo

Un resumen breve de issues de GitHub para principiantes.