Das passende Nebenläufigkeitsmodell wählen

import hashlib


def hash_file(path: str) -> str:
    """Return the SHA-256 hex digest of the file at path."""
    # Open the file in binary mode and pass it to hashlib.file_digest with "sha256".
    return ""


def hash_sequential(paths: list[str]) -> dict[str, str]:
    """Hash the files one after another: {path: digest}, in the order of paths."""
    return {}

import hashlib
import os
import tempfile

from main import hash_file, hash_sequential


def make_files(folder, contents):
    paths = []
    for number, data in enumerate(contents):
        path = os.path.join(folder, f"file{number}.bin")
        with open(path, "wb") as f:
            f.write(data)
        paths.append(path)
    return paths


def test_hash_file():
    """hash_file liefert den SHA-256-Hex-Digest der Bytes der Datei"""
    with tempfile.TemporaryDirectory() as folder:
        [path] = make_files(folder, [b"hello"])
        got = hash_file(path)
    assert got == hashlib.sha256(b"hello").hexdigest(), f"hash_file lieferte {got!r}"


def test_large_file():
    """Eine Datei von 1 MB ergibt den Digest ihrer Bytes"""
    data = bytes(range(256)) * 4096
    with tempfile.TemporaryDirectory() as folder:
        [path] = make_files(folder, [data])
        got = hash_file(path)
    assert got == hashlib.sha256(data).hexdigest(), f"hash_file lieferte {got!r}"


def test_hash_sequential():
    """hash_sequential bildet jeden Pfad auf seinen Digest ab, in der Reihenfolge von paths"""
    with tempfile.TemporaryDirectory() as folder:
        paths = make_files(folder, [b"b", b"a"])
        got = hash_sequential(paths)
    expected = [(paths[0], hashlib.sha256(b"b").hexdigest()), (paths[1], hashlib.sha256(b"a").hexdigest())]
    assert list(got.items()) == expected, f"hash_sequential lieferte {got!r}"

import hashlib
from concurrent.futures import ThreadPoolExecutor


def hash_file(path: str) -> str:
    """Return the SHA-256 hex digest of the file at path."""
    with open(path, "rb") as f:
        return hashlib.file_digest(f, "sha256").hexdigest()


def hash_sequential(paths: list[str]) -> dict[str, str]:
    """Hash the files one after another: {path: digest}, in the order of paths."""
    return {path: hash_file(path) for path in paths}


def hash_threads(paths: list[str], workers: int = 4) -> dict[str, str]:
    """Hash the files in a thread pool: file reads wait on the disk, and hashlib releases the GIL."""
    # Still one after another in the main thread. Use a ThreadPoolExecutor.
    return hash_sequential(paths)

import threading
import tempfile
import os

import main


def make_files(folder, contents):
    paths = []
    for number, data in enumerate(contents):
        path = os.path.join(folder, f"file{number}.bin")
        with open(path, "wb") as f:
            f.write(data)
        paths.append(path)
    return paths


def test_same_digests():
    """hash_threads liefert dasselbe Dict wie hash_sequential"""
    with tempfile.TemporaryDirectory() as folder:
        paths = make_files(folder, [b"one", b"two", b"three"])
        got = main.hash_threads(paths, workers=2)
        expected = main.hash_sequential(paths)
    assert list(got.items()) == list(expected.items()), f"hash_threads lieferte {got!r}"


def test_runs_in_pool_threads():
    """hash_threads ruft hash_file in Worker-Threads auf, nicht im Haupt-Thread"""
    seen = []

    def spy(path):
        seen.append(threading.current_thread().name)
        return path.upper()

    real = main.hash_file
    main.hash_file = spy
    try:
        got = main.hash_threads(["a", "b"], workers=2)
    finally:
        main.hash_file = real
    assert got == {"a": "A", "b": "B"}, f"hash_threads lieferte {got!r}"
    assert "MainThread" not in seen, f"hash_file lief in {seen!r}: nutzen Sie einen ThreadPoolExecutor"

import hashlib
from concurrent.futures import ProcessPoolExecutor, ThreadPoolExecutor


def hash_file(path: str) -> str:
    """Return the SHA-256 hex digest of the file at path."""
    with open(path, "rb") as f:
        return hashlib.file_digest(f, "sha256").hexdigest()


def hash_sequential(paths: list[str]) -> dict[str, str]:
    """Hash the files one after another: {path: digest}, in the order of paths."""
    return {path: hash_file(path) for path in paths}


def hash_threads(paths: list[str], workers: int = 4) -> dict[str, str]:
    """Hash the files in a thread pool: file reads wait on the disk, and hashlib releases the GIL."""
    with ThreadPoolExecutor(max_workers=workers) as pool:
        return dict(zip(paths, pool.map(hash_file, paths)))


def hash_processes(paths: list[str], workers: int = 4) -> dict[str, str]:
    """Hash the files in a process pool; hash_file is a module-level function, so it pickles."""
    # Still threads. Use the ProcessPoolExecutor imported above.
    return hash_threads(paths, workers)


if __name__ == "__main__":
    import sys

    print(hash_processes(sys.argv[1:]))

import tempfile
import os

import main


def make_files(folder, contents):
    paths = []
    for number, data in enumerate(contents):
        path = os.path.join(folder, f"file{number}.bin")
        with open(path, "wb") as f:
            f.write(data)
        paths.append(path)
    return paths


def test_uses_process_pool():
    """hash_processes hasht in einem ProcessPoolExecutor und liefert dasselbe Dict wie hash_sequential"""
    used = []
    real = main.ProcessPoolExecutor

    class Spy(real):
        def __init__(self, *args, **kwargs):
            used.append(kwargs.get("max_workers", args[0] if args else None))
            super().__init__(*args, **kwargs)

    main.ProcessPoolExecutor = Spy
    try:
        with tempfile.TemporaryDirectory() as folder:
            paths = make_files(folder, [b"one", b"two", b"three"])
            got = main.hash_processes(paths, workers=2)
            expected = main.hash_sequential(paths)
    finally:
        main.ProcessPoolExecutor = real
    assert used == [2], f"ProcessPoolExecutor wurde mit {used!r} erzeugt: erzeugen Sie einen mit max_workers=workers"
    assert list(got.items()) == list(expected.items()), f"hash_processes lieferte {got!r}"

import asyncio
import hashlib
from concurrent.futures import ProcessPoolExecutor, ThreadPoolExecutor


def hash_file(path: str) -> str:
    """Return the SHA-256 hex digest of the file at path."""
    with open(path, "rb") as f:
        return hashlib.file_digest(f, "sha256").hexdigest()


def hash_sequential(paths: list[str]) -> dict[str, str]:
    """Hash the files one after another: {path: digest}, in the order of paths."""
    return {path: hash_file(path) for path in paths}


def hash_threads(paths: list[str], workers: int = 4) -> dict[str, str]:
    """Hash the files in a thread pool: file reads wait on the disk, and hashlib releases the GIL."""
    with ThreadPoolExecutor(max_workers=workers) as pool:
        return dict(zip(paths, pool.map(hash_file, paths)))


def hash_processes(paths: list[str], workers: int = 4) -> dict[str, str]:
    """Hash the files in a process pool; hash_file is a module-level function, so it pickles."""
    with ProcessPoolExecutor(max_workers=workers) as pool:
        return dict(zip(paths, pool.map(hash_file, paths)))


async def hash_async(paths: list[str]) -> dict[str, str]:
    """Hash the files from asyncio: each blocking hash_file call runs in a thread via to_thread."""
    # Calling hash_file directly blocks the event loop.
    # Run each call with asyncio.to_thread, as a task in an asyncio.TaskGroup.
    return hash_sequential(paths)


if __name__ == "__main__":
    import sys

    print(hash_processes(sys.argv[1:]))

import asyncio
import tempfile
import threading
import os

import main


def make_files(folder, contents):
    paths = []
    for number, data in enumerate(contents):
        path = os.path.join(folder, f"file{number}.bin")
        with open(path, "wb") as f:
            f.write(data)
        paths.append(path)
    return paths


def test_same_digests():
    """hash_async liefert dasselbe Dict wie hash_sequential, in der Reihenfolge von paths"""
    with tempfile.TemporaryDirectory() as folder:
        paths = make_files(folder, [b"one", b"two", b"three"])
        got = asyncio.run(main.hash_async(paths))
        expected = main.hash_sequential(paths)
    assert list(got.items()) == list(expected.items()), f"hash_async lieferte {got!r}"


def test_event_loop_not_blocked():
    """hash_async führt hash_file in Worker-Threads aus, abseits der Event-Loop"""
    seen = []

    def spy(path):
        seen.append(threading.current_thread().name)
        return path.upper()

    real = main.hash_file
    main.hash_file = spy
    try:
        got = asyncio.run(main.hash_async(["a", "b"]))
    finally:
        main.hash_file = real
    assert got == {"a": "A", "b": "B"}, f"hash_async lieferte {got!r}"
    assert "MainThread" not in seen, f"hash_file lief in {seen!r}: nutzen Sie asyncio.to_thread"