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()
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