Function conflict between __get__ and __getattribute__

user_17678544 0
Viewed 47

I'm writing some code in python 3.9 and have just recently discovered about descriptors.

I have created a class which looks like this:

import threading

class OuterClass():
    class DecoratorTest():
        def __init__(self,fget,fset=None) -> None:
            self.fget=fget
            self.fset=fset
            self.event=threading.Event()
        def __getattribute__(self, name: str):
            return object.__getattribute__(self,name)
        def __get__(self,obj,objtype=None):
            return self.fget(obj)
        def __set__(self,obj,value):
            self.fset(obj,value)
            self.event.set()
            self.event.clear()
         def setter(self,fset):
            return type(self)(self.fget,fset)
      
    def __init__(self):
        self.random_value=10
    @DecoratorTest
    def test_property(self):
        return self.random_value*2
    @test_property.setter
    def test_property(self,value):
        self.random_value=value

My intent is to have a decorator which adds an event to the decorated property, so that when the property is changed, the event is set. This is an example of how I would use it:

class_iteration=OuterClass()

def test_funct():
    while True:
        class_iteration.test_property.event.wait()
        print("Changed!")

threading.Thread(target=test_funct).start()

class_iteration.test_property=100

Problem is that when I try to get event, it considers class_iteration.test_property to be an int object, thus it throws an error. I wonder if there is any way to bypass the __get__ when using __getattribute__, so that I can access the event variable, or if I need to change my approach completely.

getattribute descriptor get python
1 Answers

The problem is due to two things. One has to do with the implementation of the __get__() method in your descriptor class' implementation. It doesn't check for its obj argument being None like the one as shown in the pure Python Property class described in the documentation.

This is important because it will allow you to fix the second problem, which is with how the test_funct() thread accesses the test_property — it needs to do so through the class not the instance.
So, instead of:

class_iteration.test_property.event.wait()

it should use:

type(class_iteration).test_property.event.wait()

(Note in the code below I've changed the name of this variable from class_iteration to class_instance to better reflect what it actually is.)

This is because the threading.Event instance is shared by all instances of the OuterClass since the DecoratorTest class instance is created when its class definition is executed.

Below is your code with those fixes (and some others I made to extend the testinga little and to prevent it from hanging when it ends). I've also reformatted it to adhere to the PEP 8 - Style Guide for Python Code guidelines for improved readability.

import threading
import time


class OuterClass:
    class DecoratorTest:
        def __init__(self, fget, fset=None) -> None:
            self.fget = fget
            self.fset = fset
            self.event = threading.Event()

        def __getattribute__(self, name: str):
            return object.__getattribute__(self,name)

        def __get__(self, obj, objtype=None):
            if obj is None:  # Called for decorator instance?
                return self
            if self.fget is None:
                raise AttributeError(f'unreadable attribute {self._name}')
            return self.fget(obj)

        def __set__(self, obj, value):
            self.fset(obj, value)
            self.event.set()
            self.event.clear()

        def setter(self, fset):
            return type(self)(self.fget, fset)

    def __init__(self):
        self.random_value = 10

    @DecoratorTest
    def test_property(self):
        return self.random_value*2

    @test_property.setter
    def test_property(self,value):
        self.random_value = value


class_instance = OuterClass()

def test_funct():
    while True:
        type(class_instance).test_property.event.wait()
        print(f"Change detected: {class_instance.test_property=}")

# Daemon thread so won't prevent termination of main thread if it's running.
threading.Thread(target=test_funct, daemon=True).start()

print(f"Initial value: {class_instance.test_property=}")
time.sleep(.1)
class_instance.test_property = 100
time.sleep(.1)
class_instance.test_property = 200
time.sleep(.1)
user_3552300
Related