descriptor_demo.py

#!/usr/bin/env python
# -*- coding: utf-8 -*-
# vim: tabstop=4 shiftwidth=4 expandtab number
"""
演示Python Descriptor(描述符)的 基础语法和用例
Authors: qianweishuo<[email protected]>
Date:    2020/2/1 10:45 PM
"""


class DescRead:  # 一个`读描述符`类, 不含 __set__()方法
    def __init__(self):
        self.value = 0

    def __get__(self, instance, owner):
        return self.value


class DescWrite(DescRead):  # 一个`写描述符`类
    def __set__(self, instance, value):
        assert 0 <= value <= 9, "The value is invalid"
        self.value = value


def study(Claz):  # 针对 Claz描述的逻辑，进行预订的实验
    obj1, obj2 = Claz(), Claz()  # 新建两个Claz对象
    obj1.r1, obj1.r2 = 3, 4  # 改动obj1中的两个描述符
    print('---------- of obj1')  # 观察obj1的属性词典、它的类的属性词典、两个属性值
    show('obj1.__dict__', 'type(obj1).__dict__', 'obj1.r1', 'obj1.r2')
    print('---------- of obj2')  # 观察obj2的...
    show('obj2.__dict__', 'type(obj2).__dict__', 'obj2.r1', 'obj2.r2')
    print('---------- dict v.s. dot')  # 属性词典取值 v.s. 小数点取值语法
    show('type(obj2).__dict__["r1"]', 'type(obj2).r1')  # dot语法才会生效
    print()


import inspect


def show(*expr_args):  # 依次打印各个表达式参数的值
    # 反射机制拿到调用栈帧内的 globs/locs 变量
    caller_frame = inspect.stack()[1].frame
    globs, locs = caller_frame.f_globals, caller_frame.f_locals
    vals = []
    for e in expr_args:  # 使用 表达式、变量 字面求值
        vals.append(f"{e}\t= {repr(eval(e, globs, locs))}")
    print('\n'.join(vals))


def basic_syntax():  # 演示基本语法
    print("""实验1: 普通的类属性;
        - `obj1.r1 = 3`会直接写入 `obj1.__dict__`，用实例属性覆盖类属性，不会污染其他实例""")

    class ByClassAttrib:
        r1 = 0  # 普通的类属性
        r2 = 0

    study(ByClassAttrib)

    print("""实验2: 描述符 as 类属性;
        - 这里 r1和r2都是 类属性；因访问优先级不同，行为略有区别:
        - r1的类型是`DescRead`, 不支持`__set__()`
            - 所以`obj1.r1 = 3`会直接写入`obj1.__dict__`成为实例属性
            - 参考访问链顺序，实例属性高于`读描述符`; 所以obj1自己下次取值会得到实例属性`obj1.__dict__['r1']`，即 3
            - 而其他实例的字典`obj2.__dict__`仍然为空; 沿着访问链继续轮询到作为类属性的读描述符r1(未受污染), 即 0
        - r2的类型是`DescWrite`, 支持`__set__()`
            - 所以`ob1.r2 = 4`等价于`obj1.r2.__set__(obj2, 4)`，修改了作为类属性的r2 (而非增加实例属性);
            - 参考访问链顺序，`写描述符`高于实例属性; 所以obj1自己下次取值会咨询写描述符`obj1.r2.__get__(obj1)`，即 4
            - 其他实例也会优先轮询到`写描述符`，而这个东西是类属性(受污染)，所以 `obj2.r2 == 4`
    """)

    class ByClassDesc:
        r1 = DescRead()  # 描述符 作为 类属性
        r2 = DescWrite()

    study(ByClassDesc)

    print("""实验3: 描述符 as 实例属性 [错误用法, 略]; 无效 """)
    print()
    # class ByInstDesc:
    #     def __init__(self):
    #         self.r1 = DescRead()
    #         self.r2 = DescWrite()
    # study(ByInstDesc)

    print("""实验4: 通过描述符内部的字典, 将value与obj映射起来；可以保持__set__()能力，同时避免实例间互相污染""")
    from weakref import WeakKeyDictionary

    class DescWriteBindingDict:
        def __init__(self):
            self.value = WeakKeyDictionary()  # 可以避免实例之间互相污染
            # 如果直接用dict的话，对各实例都有强引用，阻碍GC
            # 用 WeakKeyDictionary可以缓解GC问题，但是仍然有KeyError风险

        def __get__(self, instance, owner):
            return self.value.get(instance, 0)

        def __set__(self, instance, value):
            assert 0 <= value <= 9, "The value is invalid"
            self.value[instance] = value  # 将实例 映射到 对应的value

    class ByClassDescBindingDict:
        r1 = DescRead()
        r2 = DescWriteBindingDict()

    study(ByClassDescBindingDict)

    print("""实验5: 将value直接写入obj中, 可以跟随实例一起GC""")

    class DescWriteBindingObj:
        def __init__(self, name):
            self.name = name  # 注意不需要self.value, 因为值绑定到obj上

        def __get__(self, instance, owner):
            return instance.__dict__.get(self.name, 0)  # 注意是从obj上取值

        def __set__(self, instance, value):
            assert 0 <= value <= 9, "The value is invalid"
            instance.__dict__[self.name] = value  # 值绑定到obj上

    class ByClassDescBindingObj:
        r1 = DescRead()
        r2 = DescWriteBindingObj('r2')  # 这里命名冗余

    study(ByClassDescBindingObj)

    print("""实验6: [需要 python>=3.6] 用 __set_name__() 代替 __init__() 避免命名冗余 """)
    # 对于python<3.6的情况，可以用元类语法来代替__set_name__()，也不难
    # 参见 https://lingxiankong.github.io/2014-03-28-python-descriptor.html#idmhi
    import sys

    class DescWritePy36:
        assert sys.version_info >= (3, 6), "Need python >= 3.6 for DescWritePy36.__set_name__()"

        def __set_name__(self, owner, name):
            self.name = name

        def __get__(self, instance, owner):
            return instance.__dict__.get(self.name, 0)

        def __set__(self, instance, value):
            assert 0 <= value <= 9, "The value is invalid"
            instance.__dict__[self.name] = value

    class ByClassDescPy36:
        r1 = DescRead()
        r2 = DescWritePy36()  # 这里避免了命名冗余

    study(ByClassDescPy36)


def show_usage():
    print("""\n用例1: 惰性求值""")
    import arrow
    import time

    class LazyProperty:
        def __init__(self, func):
            self.func = func
            self.name = func.__name__

        def __get__(self, instance, owner):
            value = self.func(instance)  # 调用原方法, 求出结果
            instance.__dict__[self.name] = value  # 写入原实例的属性词典中
            return instance.__dict__[self.name]

        # def __set__(self, instance, value):  # 注意实例属性高于`读描述符`,而低于`写描述符`
        #     pass # 所以，这里不能加 __set__; 空逻辑也不行
        # 如果想要保持 __set__() 能力，需要同时改写 __get__()逻辑；参考后面的例子

    class DeepThought:  # 看看类的写法，非常简洁
        @LazyProperty
        def meaning_of_life(self):
            time.sleep(1)
            return "eating"

    my_deep_thought_instance = DeepThought()
    print(arrow.now(), my_deep_thought_instance.meaning_of_life)
    print(arrow.now(), my_deep_thought_instance.meaning_of_life)  # 后两次直接读缓存
    print(arrow.now(), my_deep_thought_instance.meaning_of_life)

    print("""\n用例2: 参数检查""")
    import warnings

    INF = float('inf')

    class PowerDesc:

        def __init__(self, region=(-INF, INF)):
            self.region = region  # 取值范围

        def __set_name__(self, owner, name):  # 避免反射还能拿到变量名的骚操作
            """
            :param owner: 绑定的类型，即HParams
            :param name: 即将绑定到的变量名
            """
            self.name = name

        def __set__(self, instance, value):
            """ 写值之前，做一些检查 """
            if not self.region[0] <= value <= self.region[1]:  # 越界, 直接抛异常
                raise ValueError(f"{self.name}={value}, not within region {self.region}")
            if not (value != 0 and value & (value - 1) == 0):  # 非幂次方，打印告警
                # warnings.warn(f"{self.name}={value}, not power of 2, slow calculating")
                # warnings走的是 stderr，可能会扰乱打印顺序，这里用print便于教学演示
                print(f"{self.name}={value}, not power of 2, slow calculating")
            instance.__dict__[self.name] = value  # 写到实例的属性词典里

        def __get__(self, instance, owner):
            return instance.__dict__[self.name]  # 未设置的话，允许直接抛异常

    class HParams:
        """
        看看类的写法，变得非常简洁。
        如果改用python自带的 @property 语法，也能实现类似功能，但会麻烦很多
        - 不同属性之间难以复用逻辑
        - @property之后，还要额外为每个属性写个setter函数
        """
        image_width = PowerDesc()
        image_height = PowerDesc()
        batch_size = PowerDesc((8, 256))  # 可以通过构造参数，灵活的微调检测逻辑

    hparams = HParams()
    hparams.image_width = 1024
    hparams.image_height = 768  # 非2的幂次方, SlowCalculation
    hparams.batch_size = 32
    try:
        hparams.batch_size = 4  # 取值范围外
    except ValueError as ex:  # 此处仅演示；实际业务中不要捕获，应当正确抛出给调用者处理
        print(ex)

    print("""\n用例3: 属性监听""")

    class CallbackProperty:
        """A property that will alert observers upon updates"""

        def __init__(self, default=0):
            self.default = default
            self.callbacks = []  # 针对绑定到的类中所有实例的回调函数

        def __set_name__(self, owner, name):
            self.name = name

        def __set__(self, instance, value):
            for callback in self.callbacks:
                callback(value, instance)
            instance.__dict__[self.name] = value

        def __get__(self, instance, owner):
            if instance is None:
                return self  # 通过类访问时，直接返回描述符而非取值；以便加监听
            return instance.__dict__.get(self.name, self.default)

        def add_callback(self, callback):
            if callback in self.callbacks:
                warnings.warn(f"duplicate callback, [skip]")  # 重复的实例回调
            else:
                self.callbacks.append(callback)

    class BankAccount:  # 看看类的写法会多么简洁，与业务层逻辑完全解耦
        username: str
        balance = CallbackProperty(0)  # 短短一行，就允许业务层随意监听

        def __init__(self, username):
            self.username = username

    jack_account = BankAccount('jack')

    def low_balance_warning(value, instance: BankAccount):
        if instance.balance >= 100 > value:  # 致贫
            print(f"{instance.username} is getting poor")
        if instance.balance < 100 <= value:  # 脱贫
            print(f"{instance.username} is getting rich")

    """
    - ba.balance.add_callback(ba, low_balance_warning)  # 错误写法，ba.balance 返回int
    - 注意这里的回调逻辑都在外部，不会污染BankAccount类代码
        - BankAccount类的作者只需要将balance属性声明为 CallbackProperty即可支持该类的用户随意加监听
    - 如果改用python自带的 @property 语法，也能实现类似功能，但会复杂一些:
        - 类的作者还要实现 @balance.setter 方法
        - 类的作者要在类代码内部管理callbacks对象，导致"业务层逻辑侵入了架构层"
        - 如果有个类似于balance的其他属性，则要再写一遍类似逻辑，无法复用
    """
    BankAccount.balance.add_callback(low_balance_warning)

    jack_account.balance = 101  # 初始0 -> 101; getting rich
    jack_account.balance = 99  # 101 -> 99; getting poor
    jack_account.balance = 102  # 99 -> 102; getting rich


if __name__ == '__main__':
    basic_syntax()
    show_usage()