How to access `ApplyResult` and `Event` types in the multiprocessing library

I’ve written a working wrapper around the python multiprocessing code so I can easily start, clean up, and catch errors in my processes. I’ve recently decided to go back and add proper type hints to this code, however I can’t figure out how to use the types defined in multiprocessing correctly.

I have a function which accepts a list of ApplyResult objects and extracts those results, before returning a list of those results (if successful)

from typing import Any
from typing import TypeVar

import multiprocessing as mp

_T = TypeVar("_T")

def wait_for_pool_results(
    results: list[mp.pool.ApplyResult[_T]],
    terminate_process_event: mp.Event,
    result_timeout: int,
) -> list[_T]:
    do_stuff()

When running this code I get the following error:

    results: list[mp.pool.ApplyResult[_T]],
AttributeError: module 'multiprocessing' has no attribute 'pool'

Looking through the code, this is the location of the ApplyResult definition, and it’s not available via mp.ApplyResult either.

I could change this type hint to an Any to get around the issue (I currently do).

How do I access the ApplyResult type from python’s multiprocessing library?

Furthermore, although I can assign the mp.Event type mypy complains that Mypy: Function "multiprocessing.Event" is not valid as a type. How do I correctly access this type too?

Answer

To resolve such issues with standard library, usually typeshed repo is useful enough. In mp __init__.py Event is defined as some attribute of context. Going to mp context.py, we find out that Event is defined as synchronize.Event, and in mp synchronize.py we finally find the class definition.

The issue with mp.pool is of different kind: it has to be imported as from multiprocessing.pool import ApplyResult (or by aliasing import of multiprocessing.pool – but not as attribute of mp), because it is a nested module. See this SO question for reference.

So, the following shows proper typing in this context:

from typing import Any, TypeVar

from multiprocessing.pool import ApplyResult
from multiprocessing.synchronize import Event

_T = TypeVar("_T")

def wait_for_pool_results(
    results: list[ApplyResult[_T]],
    terminate_process_event: Event,
    result_timeout: int,
) -> list[_T]:
    ...

(playground)

Also, regarding your runtime error: if some annotations are accepted by mypy, but cause runtime issues, it often means you need from __future__ import annotations as first line of the file. It enables postponed evaluation of annotations, so any valid python is accepted in annotations.

UPD: mypy allows mp.pool attribute, because it cannot reliably track whether this import happened before. For runtime, consider the following scenario:

>>> import multiprocessing as mp

>>> mp.pool
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
AttributeError: module 'multiprocessing' has no attribute 'pool'. Did you mean: 'Pool'?

>>> from multiprocessing import pool

>>> mp.pool
<module 'multiprocessing.pool' from '...'>

So, mypy cannot reliably know whether some kind of import (including importlib.import_module or __import__ calls) of submodule happened before, so it assumes that it was. Atrribute errors of such kind are easy to resolve when they arise, so this decision makes sense from usability point of view.