Function _parallel_asyncio

scrapy/utils/asyncio.py:95–130

Execute a callable over the objects in the given iterable, in parallel, using no more than ``count`` concurrent calls. This function is only used in :meth:`scrapy.core.scraper.Scraper.handle_spider_output_async` and so it assumes that neither *callable* nor iterating *iterable* will

(
    iterable: Iterable[_T] | AsyncIterator[_T],
    count: int,
    callable_: Callable[Concatenate[_T, _P], Coroutine[Any, Any, None]],
    *args: _P.args,
    **kwargs: _P.kwargs,
)

async def _parallel_asyncio(
    iterable: Iterable[_T] | AsyncIterator[_T],
    count: int,
    callable_: Callable[Concatenate[_T, _P], Coroutine[Any, Any, None]],
    *args: _P.args,
    **kwargs: _P.kwargs,
) -> None:
    """Execute a callable over the objects in the given iterable, in parallel,
    using no more than ``count`` concurrent calls.

    This function is only used in
    :meth:`scrapy.core.scraper.Scraper.handle_spider_output_async` and so it
    assumes that neither *callable* nor iterating *iterable* will raise an
    exception.
    """
    queue: asyncio.Queue[_T | None] = asyncio.Queue(count * 2)

    async def worker() -> None:
        while True:
            item = await queue.get()
            if item is None:
                break
            try:
                await callable_(item, *args, **kwargs)
            finally:
                queue.task_done()

    async def fill_queue() -> None:
        async for item in as_async_generator(iterable):
            await queue.put(item)
        for _ in range(count):
            await queue.put(None)

    fill_task = asyncio.create_task(fill_queue())
    work_tasks = [asyncio.create_task(worker()) for _ in range(count)]
    await asyncio.wait([fill_task, *work_tasks])


class AsyncioLoopingCall: