Skip to content

omnipy.util.publisher

Observable Pydantic models for propagating attribute change notifications.

This module provides publisher models that notify subscribers when their public attributes change, including nested publisher attributes used in runtime and configuration state.

CLASS DESCRIPTION
DataPublisher

Pydantic model that publishes updates to subscribers.
RuntimeEntryPublisher

Data publisher that resets runtime subscriptions on value replacement.

DataPublisher

Bases: pyd.BaseModel


              flowchart BT
              omnipy.util.publisher.DataPublisher[DataPublisher]

                              omnipy.util.pydantic.BaseModel --> omnipy.util.publisher.DataPublisher
                


              click omnipy.util.publisher.DataPublisher href "" "omnipy.util.publisher.DataPublisher"

Pydantic model that publishes updates to subscribers.

Subscribers can watch a single public attribute or the whole model. Nested DataPublisher attributes propagate their changes to parent subscribers.

Subclassed by:
CLASS DESCRIPTION
Config

Pydantic settings for DataPublisher models.
METHOD DESCRIPTION
deepcopy

Create a deep copy with subscriptions reset on the copied object.
subscribe

Subscribe to updates for the full publisher object.
subscribe_attr

Subscribe to updates for one public attribute.
unsubscribe_all

Remove all subscriptions from this publisher and nested publishers.
Source code in src/omnipy/util/publisher.py 
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
class DataPublisher(pyd.BaseModel):
    """Pydantic model that publishes updates to subscribers.

    Subscribers can watch a single public attribute or the whole model. Nested
    ``DataPublisher`` attributes propagate their changes to parent subscribers.
    """
    class Config:
        """Pydantic settings for ``DataPublisher`` models."""
        arbitrary_types_allowed = True
        validate_assignment = True

    _self_subscriptions: list[Callable[..., None]] = pyd.PrivateAttr(default_factory=list)
    _attr_subscriptions: DefaultDict[str, list[Callable[..., None]]] = \
        pyd.PrivateAttr(default_factory=_subscribers_factory)

    def subscribe_attr(self, attr_name: str, callback_fun: Callable[..., None]):
        """Subscribe to updates for one public attribute.

        The callback is invoked immediately with the current attribute value.

        Args:
            attr_name: Public attribute name to observe.
            callback_fun: Callback receiving the current/new attribute value.

        Raises:
            AttributeError: If ``attr_name`` does not exist or is private.
        """
        if not hasattr(self, attr_name):
            raise AttributeError(f'No attribute named "{attr_name}"')
        elif attr_name.startswith('_'):
            raise AttributeError(f'Subscribing to private member "{attr_name}" not allowed')
        else:
            self._attr_subscriptions[attr_name].append(callback_fun)
            attr = getattr(self, attr_name)
            callback_fun(attr)

            if isinstance(attr, DataPublisher):
                attr.subscribe(callback_fun, do_callback=False)

    def subscribe(self, callback_fun: Callable[..., None], do_callback: bool = True) -> None:
        """Subscribe to updates for the full publisher object.

        The callback receives ``self`` whenever any public attribute changes.

        Args:
            callback_fun: Callback receiving the publisher instance.
            do_callback: When ``True``, call the callback immediately after
                registration.

        """
        self._self_subscriptions.append(callback_fun)

        def _get_attr_callback_for_publisher_child(
                attr_name: str) -> Callable[[DataPublisher], None]:
            def _attr_callback_for_publisher_child(value: object) -> None:
                callback_fun(self)

            return _attr_callback_for_publisher_child

        for attr_name in self.__class__.__fields__.keys():
            if not attr_name.startswith('_'):
                attr = getattr(self, attr_name)
                if isinstance(attr, DataPublisher):
                    attr.subscribe(
                        _get_attr_callback_for_publisher_child(attr_name), do_callback=False)

        if do_callback:
            callback_fun(self)

    def unsubscribe_all(self) -> None:
        """Remove all subscriptions from this publisher and nested publishers."""
        self._self_subscriptions.clear()
        self._attr_subscriptions.clear()

        for attr_name in self.__class__.__fields__.keys():
            if not attr_name.startswith('_'):
                attr = getattr(self, attr_name)
                if isinstance(attr, DataPublisher):
                    attr.unsubscribe_all()

    def _call_subscribers(self, attr_name: str, value: object) -> None:
        if attr_name in self._attr_subscriptions:
            for callback_fun in self._attr_subscriptions[attr_name]:
                callback_fun(value)

    def _call_self_subscribers(self) -> None:
        for callback_fun in self._self_subscriptions:
            callback_fun(self)

    def _call_all_subscribers(self, attr_name: str, value: object) -> None:
        self._call_subscribers(attr_name, value)
        self._call_self_subscribers()

    def __setattr__(self, attr_name: str, value: object) -> None:
        """Assign an attribute and publish notifications for public fields.

        Args:
            attr_name: Attribute name being assigned.
            value: New value to store.

        """
        super().__setattr__(attr_name, value)

        if not attr_name.startswith('_'):
            self._call_all_subscribers(attr_name, value)

    def deepcopy(self) -> Self:
        """Create a deep copy with subscriptions reset on the copied object.

        Returns:
            A deep-copied publisher with empty subscription registries.
        """
        self_copy = self.copy()
        self_copy._self_subscriptions = []
        self_copy._attr_subscriptions = _subscribers_factory()
        return self_copy.copy(deep=True)

Config

Pydantic settings for DataPublisher models.

ATTRIBUTE DESCRIPTION
arbitrary_types_allowed

validate_assignment

Source code in src/omnipy/util/publisher.py 
33
34
35
36
class Config:
    """Pydantic settings for ``DataPublisher`` models."""
    arbitrary_types_allowed = True
    validate_assignment = True

arbitrary_types_allowed class-attribute instance-attribute 

arbitrary_types_allowed = True

validate_assignment class-attribute instance-attribute 

validate_assignment = True

deepcopy 

deepcopy() -> Self

Create a deep copy with subscriptions reset on the copied object.

RETURNS DESCRIPTION
Self

A deep-copied publisher with empty subscription registries.
Source code in src/omnipy/util/publisher.py 
133
134
135
136
137
138
139
140
141
142
def deepcopy(self) -> Self:
    """Create a deep copy with subscriptions reset on the copied object.

    Returns:
        A deep-copied publisher with empty subscription registries.
    """
    self_copy = self.copy()
    self_copy._self_subscriptions = []
    self_copy._attr_subscriptions = _subscribers_factory()
    return self_copy.copy(deep=True)

subscribe 

subscribe(callback_fun: Callable[..., None], do_callback: bool = True) -> None

Subscribe to updates for the full publisher object.

The callback receives self whenever any public attribute changes.

PARAMETER DESCRIPTION
callback_fun

Callback receiving the publisher instance.

TYPE: Callable[..., None]

do_callback

When True, call the callback immediately after registration.

TYPE: bool DEFAULT: True

Source code in src/omnipy/util/publisher.py 
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
def subscribe(self, callback_fun: Callable[..., None], do_callback: bool = True) -> None:
    """Subscribe to updates for the full publisher object.

    The callback receives ``self`` whenever any public attribute changes.

    Args:
        callback_fun: Callback receiving the publisher instance.
        do_callback: When ``True``, call the callback immediately after
            registration.

    """
    self._self_subscriptions.append(callback_fun)

    def _get_attr_callback_for_publisher_child(
            attr_name: str) -> Callable[[DataPublisher], None]:
        def _attr_callback_for_publisher_child(value: object) -> None:
            callback_fun(self)

        return _attr_callback_for_publisher_child

    for attr_name in self.__class__.__fields__.keys():
        if not attr_name.startswith('_'):
            attr = getattr(self, attr_name)
            if isinstance(attr, DataPublisher):
                attr.subscribe(
                    _get_attr_callback_for_publisher_child(attr_name), do_callback=False)

    if do_callback:
        callback_fun(self)

subscribe_attr 

subscribe_attr(attr_name: str, callback_fun: Callable[..., None])

Subscribe to updates for one public attribute.

The callback is invoked immediately with the current attribute value.

PARAMETER DESCRIPTION
attr_name

Public attribute name to observe.

TYPE: str

callback_fun

Callback receiving the current/new attribute value.

TYPE: Callable[..., None]

RAISES DESCRIPTION
AttributeError

If attr_name does not exist or is private.
Source code in src/omnipy/util/publisher.py 
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
def subscribe_attr(self, attr_name: str, callback_fun: Callable[..., None]):
    """Subscribe to updates for one public attribute.

    The callback is invoked immediately with the current attribute value.

    Args:
        attr_name: Public attribute name to observe.
        callback_fun: Callback receiving the current/new attribute value.

    Raises:
        AttributeError: If ``attr_name`` does not exist or is private.
    """
    if not hasattr(self, attr_name):
        raise AttributeError(f'No attribute named "{attr_name}"')
    elif attr_name.startswith('_'):
        raise AttributeError(f'Subscribing to private member "{attr_name}" not allowed')
    else:
        self._attr_subscriptions[attr_name].append(callback_fun)
        attr = getattr(self, attr_name)
        callback_fun(attr)

        if isinstance(attr, DataPublisher):
            attr.subscribe(callback_fun, do_callback=False)

unsubscribe_all 

unsubscribe_all() -> None

Remove all subscriptions from this publisher and nested publishers.

Source code in src/omnipy/util/publisher.py 
 96
 97
 98
 99
100
101
102
103
104
105
def unsubscribe_all(self) -> None:
    """Remove all subscriptions from this publisher and nested publishers."""
    self._self_subscriptions.clear()
    self._attr_subscriptions.clear()

    for attr_name in self.__class__.__fields__.keys():
        if not attr_name.startswith('_'):
            attr = getattr(self, attr_name)
            if isinstance(attr, DataPublisher):
                attr.unsubscribe_all()

RuntimeEntryPublisher

Bases: DataPublisher


              flowchart BT
              omnipy.util.publisher.RuntimeEntryPublisher[RuntimeEntryPublisher]
              omnipy.util.publisher.DataPublisher[DataPublisher]

                              omnipy.util.publisher.DataPublisher --> omnipy.util.publisher.RuntimeEntryPublisher
                                omnipy.util.pydantic.BaseModel --> omnipy.util.publisher.DataPublisher
                



              click omnipy.util.publisher.RuntimeEntryPublisher href "" "omnipy.util.publisher.RuntimeEntryPublisher"
              click omnipy.util.publisher.DataPublisher href "" "omnipy.util.publisher.DataPublisher"

Data publisher that resets runtime subscriptions on value replacement.

When a public attribute is rebound to a different object and a runtime backend is attached, the runtime subscription graph is rebuilt.

Subclassed by:
Source code in src/omnipy/util/publisher.py 
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
class RuntimeEntryPublisher(DataPublisher):
    """Data publisher that resets runtime subscriptions on value replacement.

    When a public attribute is rebound to a different object and a runtime
    backend is attached, the runtime subscription graph is rebuilt.

    """

    _back: IsRuntime | None = pyd.PrivateAttr(default=None)

    def __setattr__(self, attr_name: str, value: object) -> None:
        """Assign an attribute and reset runtime subscriptions when rebinding.

        Args:
            attr_name: Attribute name being assigned.
            value: New value to store.

        """
        new_value = hasattr(self, attr_name) and getattr(self, attr_name) is not value

        super().__setattr__(attr_name, value)

        if new_value and not attr_name.startswith('_') and self._back is not None:
            self._back.reset_subscriptions()