Method _stream_sync_generator

gunicorn/dirty/worker.py:369–415

Stream chunks from a synchronous generator. Args: request_id: Request ID for the messages gen: Sync generator to iterate writer: StreamWriter for sending messages

(self, request_id, gen, writer)

            await DirtyProtocol.write_message_async(writer, response)

    async def _stream_sync_generator(self, request_id, gen, writer):
        """
        Stream chunks from a synchronous generator.

        Args:
            request_id: Request ID for the messages
            gen: Sync generator to iterate
            writer: StreamWriter for sending messages
        """
        # Sentinel value to detect end of generator
        # (StopIteration cannot be raised into a Future in Python 3.7+)
        _EXHAUSTED = object()

        def _get_next():
            try:
                return next(gen)
            except StopIteration:
                return _EXHAUSTED

        try:
            loop = asyncio.get_running_loop()
            while True:
                # Run next() in executor to avoid blocking event loop
                chunk = await loop.run_in_executor(self._executor, _get_next)
                if chunk is _EXHAUSTED:
                    break
                # Send chunk message
                await DirtyProtocol.write_message_async(
                    writer, make_chunk_message(request_id, chunk)
                )
                # Update heartbeat during long streams
                self.notify()
            # Send end message
            await DirtyProtocol.write_message_async(
                writer, make_end_message(request_id)
            )
        except Exception as e:
            # Error during streaming - send error message
            tb = traceback.format_exc()
            self.log.error("Error during streaming: %s\n%s", e, tb)
            response = make_error_response(
                request_id,
                DirtyAppError(str(e), traceback=tb)
            )
            await DirtyProtocol.write_message_async(writer, response)
        finally:
            gen.close()

    async def _stream_async_generator(self, request_id, gen, writer):
        """