MobX reinvent in Python

pymobx.py

# observable
# observer
# reaction
# derivation
import time
from typing import Callable
from collections import defaultdict


from attr import attrib, attrs


access_observables = []
derivation_graphs = defaultdict(list)

class Observable:

    def __getattribute__(self, attr):
        # track attr
        if not attr.startswith('_'):
            access_observables.append(attr)
        return super(Observable, self).__getattribute__(attr)

    def __setattr__(self, attr, value):
        super(Observable, self).__setattr__(attr, value)

        # trigger on every change
        if not attr.startswith('_'):
            for runner in derivation_graphs[attr]:
                runner()






@attrs(kw_only=True)
class Album(Observable):

    title: str = attrib()
    year: int = attrib()
    play_count: int = attrib(default=0)


def create_reaction(on_change):
    def f(track):
        global access_observables
        access_observables = []
        track() # track function 
        for attr in access_observables: # on_change handler, callback
            derivation_graphs[attr].append(on_change)
    return f



def autorun(callback:Callable):
    # run once
    # build derivation graph
    cb = create_reaction(callback)
    cb(callback)



def main():
    album = Album(title="OK", year=1997)
    autorun(lambda: print(f"**play count: {album.play_count}"))

    album.title

    autorun(lambda: print(f"**album title: {album.title}"))
    album.title = '1'

    x = 1
    while x <= 3:
        time.sleep(x)
        album.play_count = x
        # album.title = f'title {x}'
        x += 1


if __name__ == "__main__":
    main()