Function schedule_formatting

src/black/concurrency.py:142–221

Run formatting of `sources` in parallel using the provided `executor`. (Use ProcessPoolExecutors for actual parallelism.) `write_back`, `fast`, and `mode` options are passed to :func:`format_file_in_place`.

(
    sources: set[Path],
    fast: bool,
    write_back: WriteBack,
    mode: Mode,
    report: Report,
    loop: asyncio.AbstractEventLoop,
    executor: Executor,
    no_cache: bool = False,
)

async def schedule_formatting(
    sources: set[Path],
    fast: bool,
    write_back: WriteBack,
    mode: Mode,
    report: Report,
    loop: asyncio.AbstractEventLoop,
    executor: Executor,
    no_cache: bool = False,
) -> None:
    """Run formatting of `sources` in parallel using the provided `executor`.

    (Use ProcessPoolExecutors for actual parallelism.)

    `write_back`, `fast`, and `mode` options are passed to
    :func:`format_file_in_place`.
    """
    cache = None if no_cache else Cache.read(mode)
    if cache is not None and write_back not in (
        WriteBack.DIFF,
        WriteBack.COLOR_DIFF,
    ):
        sources, cached = cache.filtered_cached(sources)
        for src in sorted(cached):
            report.done(src, Changed.CACHED)
    if not sources:
        return

    cancelled = []
    sources_to_cache = []
    lock = None
    manager = None
    if write_back in (WriteBack.DIFF, WriteBack.COLOR_DIFF):
        # For diff output, we need locks to ensure we don't interleave output
        # from different processes.
        manager = Manager()
        lock = manager.Lock()

    try:
        tasks = {
            asyncio.ensure_future(
                loop.run_in_executor(
                    executor, format_file_in_place, src, fast, mode, write_back, lock
                )
            ): src
            for src in sorted(sources)
        }
        pending = tasks.keys()
        try:
            loop.add_signal_handler(signal.SIGINT, cancel, pending)
            loop.add_signal_handler(signal.SIGTERM, cancel, pending)
        except NotImplementedError:
            # There are no good alternatives for these on Windows.
            pass
        while pending:
            done, _ = await asyncio.wait(pending, return_when=asyncio.FIRST_COMPLETED)
            for task in done:
                src = tasks.pop(task)