Search code examples
pythonpython-multiprocessingpython-typing

Type hints for a function (multiprocessing)


The following function creates an instance of a class that can be shared between processes:

def create_shared_object (constructor: ?) -> ?:

    from multiprocessing.managers import BaseManager;

    BaseManager.register('name', constructor);

    manager = BaseManager();

    manager.start();

    return manager.name();

What are the correct type hints?


Solution

  • Such generated methods of a manager returns an instance of the BaseProxy type.

    Depending on the argument to proxytype, the type might change:

    def create_shared_object(...):
        BaseManager.register(..., proxytype = CustomProxy)
        ...
    
    class C: ...
    
    t = type(create_shared_object(C))
    print(t)  # CustomProxy
    

    When proxytype is not given or None, an AutoProxy class is dynamicaly generated. It is not importable and cannot be used as a type hint (in which case, use BaseProxy).

    def create_shared_object(...):
        BaseManager.register(..., proxytype = None)
    
    ...
    
    # multiprocessing.managers.AutoProxy[name]
    print(t)
    
    # (AutoProxy[name], BaseProxy, object)
    print(t.__mro__)
    

    As for constructor, the type BaseManager.register expects is Callable[..., object] | None. Classes are also callable, so it is fine to use this type even if you intend to pass a class.

    # From typeshed
    class BaseManager:
        ...
        @classmethod
        def register(
            cls,
            typeid: str,
            callable: Callable[..., object] | None = None,
            #         ^^^^^^^^^^^^^^^^^^^^^^^^^^^^
            proxytype: Any = None,
            exposed: Sequence[str] | None = None,
            method_to_typeid: Mapping[str, str] | None = None,
            create_method: bool = True,
        ) -> None: ...