Search code examples
pythonpython-3.xmultiprocessing

How to use @property with Python ProcessManager


I have a class with a @property in it and want to raise a RuntimeError if the property is accessed before the function is set from another function. My problem is now that when I use this class in a ProcessManager, the property is automatically called when building this manager, leading to the error being raised all the time. How can I get around this?

My Code is the Following:

from multiprocessing.managers import BaseManager
class TestProvider:
    def __init__(self):
        self.__result = None

    def calculate(self, i1, i2):
        self.__result = i1 + i2

    @property
    def result(self):
        if not self.__result:
            raise RuntimeError("Tried to access property which is Not calculated")
        return self.__result



if __name__ == "__main__":
    BaseManager.register("TestProvider", TestProvider)
    manager = BaseManager()
    manager.start()

    provider = manager.TestProvider()

Solution

  • Looks like managers don't support objects with properties. There are two problems, the first is to prevent the manager from trying to invoke the property getter. this could be done by specifying the exposed methods when registering the type and just not listing this property.

    BaseManager.register("TestProvider", TestProvider, exposed=["calculate"])
    

    which leads to the second problem, how do we actually access this property ? one way is to provide a get_result method that you expose instead.

    another method is to provide your own proxy that implements this property, which behaves the same as the original object.

    from multiprocessing.managers import BaseManager, BaseProxy
    class TestProvider:
        def __init__(self):
            self.__result = None
    
        def calculate(self, i1, i2):
            self.__result = i1 + i2
    
        @property
        def result(self):
            if not self.__result:
                raise RuntimeError("Tried to access property which is Not calculated")
            return self.__result
    
    class TestProviderProxy(BaseProxy):
        _exposed_ = ['calculate', '__getattribute__']
    
        def calculate(self, *args):
            self._callmethod('calculate', args)
    
        @property
        def result(self):
            return self._callmethod("__getattribute__", ["result"])
    
    if __name__ == "__main__":
        BaseManager.register("TestProvider", TestProvider, 
                             proxytype=TestProviderProxy)
        manager = BaseManager()
        manager.start()
    
        provider = manager.TestProvider()
        provider.calculate(1,2)
        print(provider.result)
    
    3