State sharing between Python and Javascript

[r0x0r]

The next major version of pywebview will introduce state - a way to seamlessly share data between Python and Javascript. State is represented by the pywebview.state object in Javascript and Window.state object in Python. Updating a property on either object will update its value on the other side. An example can be found here.

Additionally, each state object emits an event when state is updated. Implementation details are not set in stone, but currently the syntax is pywebview.state.addEventHandler('change', handler) and window.state += lambda event_type, key, value: pass

State is implemented by means of a proxy in Javascript and an observable class in Python.

The current ahem state can be tracked in the store branch

Ideas, suggestions or questions are welcomed

[r0x0r]

As far as I am concerned, basic implementation is done now. Latest changes are pushed into the 6.0 branch.

One limitation of the current version is that it detects changes only on the window.state object itself and mutations of the objects are not registered. Ie this will trigger a state update window.state.counter = 1 and this will not

window.state.nested_counter = {
   'counter': 0
}
window.state.nested_counter['counter'] = 1 # this goes undetected

I have attempted to solve this with two different approaches

1. By converting dicts and lists to observable versions, which detect and propagate changes. The implementation turned out rather complicated and I could not get it fully working. Besides I am not exactly sure if it is a good approach to convert user data be default
2. By making State object assignable and introducing sub-states. Something like this

sub_state = State({'counter': 0})
window.state.nested_counter = sub_state
sub_state.counter = 1 # this will be detected

Sub-states have exactly same API as the main state

Anyhow neither of these approaches are implemented, nor will be for the 6.0. It remains to be seen if this feature is needed.

Another feature that might be introduced in the future is global state, a state object which can be shared between different windows.

[github-actions[bot]]

This issue has been automatically marked as stale because it has not had recent activity. It will be closed if no further activity occurs. Thank you for your contributions.

[jeanbaptiste444]

Hello, Thank you for this feature. Is there a release date for 6.0 ?

[r0x0r]

@jeanbaptiste444 sometime later this year. There are a few things left to be implemented.

[github-actions[bot]]

This issue has been automatically marked as stale because it has not had recent activity. It will be closed if no further activity occurs. Thank you for your contributions.

[CyberQin]

Does Traitlets package -> Observe feature help this? Or a self made class ObservableDict

from collections.abc  import MutableSequence, MutableSet
import copy 
 
class ObservableContainer:
    def __init__(self, *args, **kwargs):
        self._callbacks = []
        super().__init__(*args, **kwargs)
        
    def add_callback(self, callback):
        """Add a change notification callback
        添加变更通知回调函数
        """
        self._callbacks.append(callback) 
 
    def remove_callback(self, callback):
        """Remove a callback
        移除回调函数
        """
        if callback in self._callbacks:
            self._callbacks.remove(callback) 
 
    def _notify(self, path=None, operation=None, key=None, old_value=None, new_value=None):
        """Notify all registered callbacks
        通知所有已注册的回调函数
        """
        path = path or []
        for callback in self._callbacks:
            callback(path, operation, key, old_value, new_value)
 
class ObservableDict(ObservableContainer, dict):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self._wrap_nested()
 
    def _wrap_nested(self):
        """Convert nested containers to observable versions
        将嵌套容器转换为可观察版本
        """
        for key, value in self.items(): 
            self._make_observable(key, value)
 
    def _make_observable(self, key, value):
        """Convert a value to observable version if needed
        如果需要，将值转换为可观察版本
        """
        if isinstance(value, dict) and not isinstance(value, ObservableDict):
            self[key] = ObservableDict(value)
            self[key].add_callback(
                lambda p, op, k, ov, nv, parent_key=key: 
                self._notify([parent_key] + p, op, k, ov, nv)
            )
        elif isinstance(value, (list, tuple)) and not isinstance(value, ObservableList):
            self[key] = ObservableList(value)
            self[key].add_callback(
                lambda p, op, k, ov, nv, parent_key=key: 
                self._notify([parent_key] + p, op, k, ov, nv)
            )
        elif isinstance(value, set) and not isinstance(value, ObservableSet):
            self[key] = ObservableSet(value)
            self[key].add_callback(
                lambda p, op, k, ov, nv, parent_key=key: 
                self._notify([parent_key] + p, op, k, ov, nv)
            )
 
    def __setitem__(self, key, value):
        old_value = self.get(key,  None)
        self._make_observable(key, value)
        super().__setitem__(key, value)
        self._notify([], 'set', key, old_value, value)
 
    def __delitem__(self, key):
        old_value = self.get(key,  None)
        super().__delitem__(key)
        self._notify([], 'del', key, old_value, None)
 
class ObservableList(ObservableContainer, MutableSequence):
    def __init__(self, iterable=None):
        super().__init__()
        self._list = list(iterable) if iterable else []
        self._wrap_nested()
 
    def _wrap_nested(self):
        """Convert nested containers to observable versions
        将嵌套容器转换为可观察版本
        """
        for i, value in enumerate(self._list):
            self._make_observable(i, value)
 
    def _make_observable(self, index, value):
        """Convert a value to observable version if needed
        如果需要，将值转换为可观察版本
        """
        if isinstance(value, dict) and not isinstance(value, ObservableDict):
            self._list[index] = ObservableDict(value)
            self._list[index].add_callback(
                lambda p, op, k, ov, nv, parent_index=index: 
                self._notify([parent_index] + p, op, k, ov, nv)
            )
        elif isinstance(value, (list, tuple)) and not isinstance(value, ObservableList):
            self._list[index] = ObservableList(value)
            self._list[index].add_callback(
                lambda p, op, k, ov, nv, parent_index=index: 
                self._notify([parent_index] + p, op, k, ov, nv)
            )
        elif isinstance(value, set) and not isinstance(value, ObservableSet):
            self._list[index] = ObservableSet(value)
            self._list[index].add_callback(
                lambda p, op, k, ov, nv, parent_index=index: 
                self._notify([parent_index] + p, op, k, ov, nv)
            )
 
    def __getitem__(self, index):
        return self._list[index]
 
    def __setitem__(self, index, value):
        old_value = copy.deepcopy(self._list[index])  if index < len(self._list) else None
        self._make_observable(index, value)
        self._list[index] = value
        self._notify([], 'set', index, old_value, value)
 
    def __delitem__(self, index):
        old_value = copy.deepcopy(self._list[index]) 
        del self._list[index]
        self._notify([], 'del', index, old_value, None)
 
    def __len__(self):
        return len(self._list)
 
    def insert(self, index, value):
        self._make_observable(index, value)
        self._list.insert(index,  value)
        self._notify([], 'insert', index, None, value)
 
    def clear(self):
        """Remove all items from list with change notification
        清空列表并发送变更通知
        """
        old_list = copy.deepcopy(self._list) 
        self._list.clear() 
        self._notify([], 'clear', None, old_list, [])
 
class ObservableSet(ObservableContainer, MutableSet):
    def __init__(self, iterable=None):
        """Initialize observable set with optional iterable
        使用可选的可迭代对象初始化可观察集合
        """
        self._set = set(iterable) if iterable else set()
        super().__init__()
 
    def __contains__(self, value):
        """Check if value exists in set
        检查值是否存在于集合中
        """
        return value in self._set 
 
    def __iter__(self):
        """Get iterator for the set
        获取集合的迭代器
        """
        return iter(self._set)
 
    def __len__(self):
        """Get the size of the set
        获取集合的大小
        """
        return len(self._set)
  
    def add(self, value):
        if value not in self._set:
            self._set.add(value) 
            self._notify([], 'add', None, None, value)
  
    def discard(self, value):
        if value in self._set:
            self._set.discard(value) 
            self._notify([], 'discard', None, value, None)
     
if __name__ == '__main__':
    def change_handler(path, operation, key, old_value, new_value):
        print(f"Change detected at path {path}:")
        print(f"  Operation: {operation}")
        if key is not None:
            print(f"  Key/Index: {key}")
        if old_value is not None:
            print(f"  Old value: {old_value}")
        if new_value is not None:
            print(f"  New value: {new_value}")
        print("---")
    
    data = ObservableDict({
        'name': 'Alice',
        'scores': [85, 90, 78],
        'details': {
            'age': 30,
            'hobbies': ['reading', 'swimming']
        }
    })
    
    data.add_callback(change_handler) 
    
    # 这些操作现在都能正常工作 
    data['scores'].append(95)  # 列表操作 
    data['details']['age'] = 31  # 字典操作
    data['details']['hobbies'].pop()  # 嵌套列表操作

The tested export like:

(.venv) PS E:\test_observe> python .\ObservableDict.py
Change detected at path ['scores']:
  Operation: insert
  Key/Index: 3
  New value: 95
---
Change detected at path ['details']:
  Operation: set
  Key/Index: age
  Old value: 30
  New value: 31
---
Change detected at path ['details', 'hobbies']:
  Operation: del
  Key/Index: -1
  Old value: swimming
---

[CyberQin]

or just claim exactly in docs that: variable in pywebview.state needs to assignment to new value (like javascript vuejs framework did before)

[r0x0r]

@CyberQin state implementation is done for the time being. You can play with it in the 6.0 branch

[r0x0r]

6.0 is released now.