We can't find the internet
Attempting to reconnect
Something went wrong!
Attempting to reconnect
← Seams, Modules, Packaging and Tooling step 15 of 36
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
executorisNone, create aThreadPoolExecutor(max_workers=workers)and setowns_executor = True. -
If an
executoris supplied, store it and setowns_executor = False. -
__exit__shuts the executor down only whenowns_executoris true. -
runmapsfnoveritemson the executor and returns the results in input order. -
Both
executorandowns_executorare 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]:
-
If
provide_executor, build aThreadPoolExecutor(max_workers=workers)yourself; otherwise passexecutor=None. -
Build a
BatchProcessor[int, str]over aformat_item(n: int) -> strthat returnsf"#{n}", and runvaluesthrough it inside awithblock. -
After the block, probe the processor’s executor:
processor.executor.submit(len, "abcd").result(). If it still accepts work the answer is4; if it has been shut down the call raisesRuntimeErrorand the answer is-1. - Shut down the executor you created yourself, if any.
-
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__.
Stuck?
Python reference solution
Sign in to attempt this problem and reveal the reference solution.