Commit 82d36a2f authored by Yakubov, Sergey's avatar Yakubov, Sergey
Browse files

add types

parent 373ca67f
Loading
Loading
Loading
Loading
Loading
+1 −1
Original line number Diff line number Diff line
[tool.poetry]
name = "mvvm-lib"
version = "0.3.0"
version = "0.3.1"
description = "A Python Package for Model-View-ViewModel pattern"
authors = ["Yakubov, Sergey <yakubovs@ornl.gov>"]
readme = "README.md"
+82 −13
Original line number Diff line number Diff line
"""Abstract interfaces."""
"""Abstract interfaces and type definitions."""

import functools
from abc import ABC, abstractmethod
from typing import Any
from typing import Any, Callable, Dict, Optional, Union

LinkedObjectType = Optional[Union[object, Dict[str, Any], Callable]]
LinkedObjectAttributesType = Optional[list[str]]
CallbackAfterUpdateType = Optional[Callable[[Optional[str]], None]]
ConnectCallbackType = Union[None, Callable[[Any, Optional[str]], None]]


class Communicator(ABC):
    """Abstract communicator class.

    Provides methods required for binding to communicate between ViewModel and View.
    """

    @abstractmethod
    def connect(self, connector: Any = None) -> ConnectCallbackType:
        """
        Connect a GUI element to a linked object.

        This method should be called from the View side to establish a
        connection between a GUI element and a linked object, which
        is passed during the bind creation from the ViewModel side.

        Parameters
        ----------
        connector : Any, optional
            The GUI element or object to connect. None can be used in some implementations
            to indicate that GUI element(s) should be automatically identified using the linked object.

        Returns
        -------
        Union[None, Callable]
        Depending on the specific implementation, returns None or a callback function
        that can be used to update the linked object.
        """
        raise Exception("Please implement in a concrete class")

    @abstractmethod
    def update_in_view(self, value: Any) -> None:
        """
        Update UI component(s) with the provided value.

        Parameters
        ----------
        value : Any
            The new value to be reflected in the view.

        Returns
        -------
        None
            This method does not return a value.
        """
        raise Exception("Please implement in a concrete class")


class BindingInterface(ABC):
@@ -10,18 +61,36 @@ class BindingInterface(ABC):

    @abstractmethod
    def new_bind(
        self, linked_object: Any = None, linked_object_arguments: Any = None, callback_after_update: Any = None
    ) -> Any:
        raise Exception("Please implement in a concrete class")
        self,
        linked_object: LinkedObjectType = None,
        linked_object_arguments: LinkedObjectAttributesType = None,
        callback_after_update: CallbackAfterUpdateType = None,
    ) -> Communicator:
        """
        Bind a ViewModel or Model variable to a GUI framework element, allowing for synchronized updates.

        This method creates a binding between a variable in a ViewModel/Model and a corresponding element
        in the GUI.

def rsetattr(obj: Any, attr: str, val: Any) -> None:
    pre, _, post = attr.rpartition(".")
    return setattr(rgetattr(obj, pre) if pre else obj, post, val)
        Parameters
        ----------
        linked_object : object, dictionary or function, optional
            Instance to link with the ViewModel/Model variable. When specified, changes in View
            trigger update for this instance.

        linked_object_arguments : list of str, optional
            If the `linked_object` is a class instance(object) one can provide a list of argument names associated
            with `linked_object` that define specific attributes
            to bind. If not provided, the default behavior is to bind all attributes.

def rgetattr(obj: Any, attr: str, *args: Any) -> Any:
    def _getattr(obj: Any, attr: str) -> Any:
        return getattr(obj, attr, *args)
        callback_after_update : Callable, optional
            A function to be called after each update to `linked_object`. Useful for additional
            actions or processing post-update.

    return functools.reduce(_getattr, [obj] + attr.split("."))
        Returns
        -------
        Communicator
            An object that manages the binding, allowing updates to propagate between the
            ViewModel/Model variable and the GUI framework element.
        """
        raise Exception("Please implement in a concrete class")
+2 −1
Original line number Diff line number Diff line
@@ -5,7 +5,8 @@ from typing import Any

import param

from ..interface import BindingInterface, rgetattr, rsetattr
from ..interface import BindingInterface
from ..utils import rgetattr, rsetattr


def is_parameterized(var: Any) -> bool:
+5 −3
Original line number Diff line number Diff line
@@ -2,6 +2,8 @@

from typing import Any, Optional

from ..utils import rsetattr

try:
    from PyQt6.QtCore import QObject, pyqtSignal
except Exception:
@@ -10,14 +12,14 @@ except Exception:

import inspect

from ..interface import BindingInterface, rsetattr
from ..interface import BindingInterface


def is_callable(var: Any) -> bool:
    return inspect.isfunction(var) or inspect.ismethod(var)


class Communicate(QObject):
class Communicator(QObject):
    """Communicator class, that provides methods required for binding to communicate between ViewModel and View."""

    signal = pyqtSignal(object)
@@ -72,4 +74,4 @@ class PyQtBinding(BindingInterface):
        For PyQt we use pyqtSignal to trigger GU
        I update and linked_object to trigger ViewModel/Model update
        """
        return Communicate(linked_object, linked_object_arguments, callback_after_update)
        return Communicator(linked_object, linked_object_arguments, callback_after_update)
+44 −34
Original line number Diff line number Diff line
@@ -2,11 +2,20 @@

import asyncio
import inspect
from typing import Any, Optional, Union
from typing import Any, Callable, Optional, Union, cast

from trame_server.state import State
from typing_extensions import override

from ..interface import BindingInterface, rgetattr, rsetattr
from ..interface import (
    BindingInterface,
    CallbackAfterUpdateType,
    Communicator,
    ConnectCallbackType,
    LinkedObjectAttributesType,
    LinkedObjectType,
)
from ..utils import rgetattr, rsetattr


def is_async() -> bool:
@@ -21,15 +30,15 @@ def is_callable(var: Any) -> bool:
    return inspect.isfunction(var) or inspect.ismethod(var)


class Communicator:
    """Communicator class, that provides methods required for binding to communicate between ViewModel and View."""
class TrameCommunicator(Communicator):
    """Communicator implementation for Trame."""

    def __init__(
        self,
        state: Any,
        viewmodel_linked_object: Any = None,
        linked_object_attributes: Any = None,
        callback_after_update: Any = None,
        state: State,
        viewmodel_linked_object: LinkedObjectType = None,
        linked_object_attributes: LinkedObjectAttributesType = None,
        callback_after_update: CallbackAfterUpdateType = None,
    ) -> None:
        self.state = state
        self.viewmodel_linked_object = viewmodel_linked_object
@@ -37,28 +46,28 @@ class Communicator:
        self.viewmodel_callback_after_update = callback_after_update
        self.connection: Union[CallBackConnection, StateConnection]

    def _set_linked_object_attributes(self, linked_object_attributes: Any, viewmodel_linked_object: Any) -> None:
        self.linked_object_attributes = None
    def _set_linked_object_attributes(
        self, linked_object_attributes: LinkedObjectAttributesType, viewmodel_linked_object: LinkedObjectType
    ) -> None:
        self.linked_object_attributes: LinkedObjectAttributesType = None
        if (
            viewmodel_linked_object
            and not isinstance(viewmodel_linked_object, dict)
            and not is_callable(viewmodel_linked_object)
        ):
            if not linked_object_attributes:
                self.linked_object_attributes = {
                    k: v for k, v in viewmodel_linked_object.__dict__.items() if not k.startswith("_")
                }
                self.linked_object_attributes = [
                    k for k in viewmodel_linked_object.__dict__.keys() if not k.startswith("_")
                ]
            else:
                self.linked_object_attributes = linked_object_attributes

    def connect(self, connector: Any = None) -> Any:
        # connect should be called from View side to connect a
        # GUI element (via it's name in Trame state object)
        # and a linked_object (passed during bind creation from ViewModel side)
    @override
    def connect(self, connector: Any = None) -> ConnectCallbackType:
        if is_callable(connector):
            self.connection = CallBackConnection(self, connector)
        else:
            self.connection = StateConnection(self, connector)
            self.connection = StateConnection(self, str(connector) if connector else None)
        return self.connection.get_callback()

    def update_in_view(self, value: Any) -> None:
@@ -68,7 +77,7 @@ class Communicator:
class CallBackConnection:
    """Connection that uses callback."""

    def __init__(self, communicator: Communicator, callback: Any) -> None:
    def __init__(self, communicator: TrameCommunicator, callback: Callable[[Any], None]) -> None:
        self.callback = callback
        self.communicator = communicator
        self.viewmodel_linked_object = communicator.viewmodel_linked_object
@@ -82,7 +91,7 @@ class CallBackConnection:
            else:
                self.viewmodel_linked_object.update({key: value})
        elif is_callable(self.viewmodel_linked_object):
            self.viewmodel_linked_object(value)
            cast(Callable, self.viewmodel_linked_object)(value)
        elif isinstance(self.viewmodel_linked_object, object):
            if not key:
                raise Exception("Cannot update", self.viewmodel_linked_object, ": key is missing")
@@ -96,14 +105,14 @@ class CallBackConnection:
    def update_in_view(self, value: Any) -> None:
        self.callback(value)

    def get_callback(self) -> Any:
    def get_callback(self) -> ConnectCallbackType:
        return self._update_viewmodel_callback


class StateConnection:
    """Connection that uses a state variable."""

    def __init__(self, communicator: Communicator, state_variable_name: str) -> None:
    def __init__(self, communicator: TrameCommunicator, state_variable_name: Optional[str]) -> None:
        self.state_variable_name = state_variable_name
        self.communicator = communicator
        self.state = communicator.state
@@ -112,7 +121,7 @@ class StateConnection:
        self.linked_object_attributes = communicator.linked_object_attributes
        self._connect()

    def _on_state_update(self, attribute_name: str, name_in_state: str) -> Any:
    def _on_state_update(self, attribute_name: str, name_in_state: str) -> Callable:
        def update(**_kwargs: Any) -> None:
            rsetattr(self.viewmodel_linked_object, attribute_name, self.state[name_in_state])
            if self.viewmodel_callback_after_update:
@@ -152,14 +161,14 @@ class StateConnection:
                    name_in_state = self._get_name_in_state(attribute_name)
                    f = self._on_state_update(attribute_name, name_in_state)
                    self.state.change(name_in_state)(f)
            else:
            elif state_variable_name:

                @self.state.change(state_variable_name)
                def update_viewmodel_callback(**kwargs: Any) -> None:
                def update_viewmodel_callback(**kwargs: dict) -> None:
                    if isinstance(self.viewmodel_linked_object, dict):
                        self.viewmodel_linked_object.update(kwargs[state_variable_name])
                    elif is_callable(self.viewmodel_linked_object):
                        self.viewmodel_linked_object(kwargs[state_variable_name])
                        cast(Callable, self.viewmodel_linked_object)(kwargs[state_variable_name])
                    else:
                        raise Exception("cannot update", self.viewmodel_linked_object)
                    if self.viewmodel_callback_after_update:
@@ -171,10 +180,10 @@ class StateConnection:
                name_in_state = self._get_name_in_state(attribute_name)
                value_to_change = rgetattr(value, attribute_name)
                self._set_variable_in_state(name_in_state, value_to_change)
        else:
        elif self.state_variable_name:
            self._set_variable_in_state(self.state_variable_name, value)

    def get_callback(self) -> Any:
    def get_callback(self) -> ConnectCallbackType:
        return None


@@ -184,10 +193,11 @@ class TrameBinding(BindingInterface):
    def __init__(self, state: State) -> None:
        self._state = state

    @override
    def new_bind(
        self, linked_object: Any = None, linked_object_arguments: Any = None, callback_after_update: Any = None
    ) -> Communicator:
        # each new_bind returns an object that can be used to bind a ViewModel/Model variable
        # with a corresponding GUI framework element
        # for Trame we use state to trigger GUI update and linked_object to trigger ViewModel/Model update
        return Communicator(self._state, linked_object, linked_object_arguments, callback_after_update)
        self,
        linked_object: LinkedObjectType = None,
        linked_object_arguments: LinkedObjectAttributesType = None,
        callback_after_update: CallbackAfterUpdateType = None,
    ) -> TrameCommunicator:
        return TrameCommunicator(self._state, linked_object, linked_object_arguments, callback_after_update)
Loading