Function wait_for_all_tasks_for_finish

src/crawlee/_utils/wait.py:49–86

Wait for all tasks to finish or until the timeout is reached. Args: tasks: A sequence of asyncio tasks to wait for. logger: Logger to use for reporting. timeout: How long should we wait before cancelling the tasks.

(
    tasks: Sequence[asyncio.Task],
    *,
    logger: Logger,
    timeout: timedelta | None = None,
)

async def wait_for_all_tasks_for_finish(
    tasks: Sequence[asyncio.Task],
    *,
    logger: Logger,
    timeout: timedelta | None = None,
) -> None:
    """Wait for all tasks to finish or until the timeout is reached.

    Args:
        tasks: A sequence of asyncio tasks to wait for.
        logger: Logger to use for reporting.
        timeout: How long should we wait before cancelling the tasks.
    """
    if not tasks:
        return

    timeout_secs = timeout.total_seconds() if timeout else None
    try:
        _, pending = await asyncio.wait(tasks, timeout=timeout_secs)
        if pending:
            logger.warning('Waiting timeout reached; canceling unfinished tasks.')
    except asyncio.CancelledError:
        logger.warning('Asyncio wait was cancelled; canceling unfinished tasks.')
        raise
    finally:
        for task in tasks:
            if not task.done():
                task.cancel()
                with suppress(asyncio.CancelledError):
                    await task
            # If task is done, access the result to clear any exceptions
            else:
                try:
                    task.result()
                except asyncio.CancelledError:
                    pass
                except Exception as e:
                    logger.warning(f'Task raised an exception: {e}')