Skip to content

← Seams, Modules, Packaging and Tooling step 15 of 36

Medium End-to-End

Own what you create, borrow what you are given

Write a library-quality class that can be dropped into somebody else’s composition root without wrecking it.

class BatchProcessor[T, R]:
    def __init__(
        self,
        fn: Callable[[T], R],
        *,
        executor: Executor | None = None,
        workers: int | None = None,
    ) -> None: ...

    owns_executor: bool
    executor: Executor

    def __enter__(self) -> Self: ...
    def __exit__(self, exc_type, exc, tb) -> None: ...
    def run(self, items: Iterable[T]) -> list[R]: ...

Rules:

  • If executor is None, create a ThreadPoolExecutor(max_workers=workers) and set owns_executor = True.
  • If an executor is supplied, store it and set owns_executor = False.
  • __exit__ shuts the executor down only when owns_executor is true.
  • run maps fn over items on the executor and returns the results in input order.
  • Both executor and owns_executor are public attributes; the harness inspects them.

Then the graded entrypoint:

def solve(values: list[int], provide_executor: bool, workers: int) -> tuple[list[str], bool, int]:
  1. If provide_executor, build a ThreadPoolExecutor(max_workers=workers) yourself; otherwise pass executor=None.
  2. Build a BatchProcessor[int, str] over a format_item(n: int) -> str that returns f"#{n}", and run values through it inside a with block.
  3. After the block, probe the processor’s executor: processor.executor.submit(len, "abcd").result(). If it still accepts work the answer is 4; if it has been shut down the call raises RuntimeError and the answer is -1.
  4. Shut down the executor you created yourself, if any.
  5. Return (results, processor.owns_executor, probe).

What the probe is testing. A borrowed executor must still be alive: the caller may be sharing it with three other components and disposing of it at the end of their main(). A library that shuts down an executor it did not create is not being tidy — it is reaching into another program’s composition root and breaking it, from inside a with block that looked harmless. An executor you created yourself must be dead, or you have leaked threads for the life of the process.

Note what is not in this design: no set_start_method, no pool created at import time, no module-level singleton. A library that does any of those takes a decision away from its users that they cannot take back.

Returns a 3-tuple, not a list. The harness compares container types exactly.

Your submission must pass mypy --strict. Use PEP 695 syntax (class BatchProcessor[T, R]:), annotate __exit__ properly (type[BaseException] | None, BaseException | None, TracebackType | None), and return Self from __enter__.