Skip to content

omnipy.shared.protocols.util

Utility protocols shared across Omnipy runtime and configuration code.

CLASS DESCRIPTION
IsDataPublisher

Protocol for objects that publish change notifications to subscribers.
IsDataclass

Protocol for objects that expose the standard dataclass marker fields.

IsDataPublisher

Bases: Protocol


              flowchart BT
              omnipy.shared.protocols.util.IsDataPublisher[IsDataPublisher]

              

              click omnipy.shared.protocols.util.IsDataPublisher href "" "omnipy.shared.protocols.util.IsDataPublisher"

Protocol for objects that publish change notifications to subscribers.

Subclassed by:
METHOD DESCRIPTION
deepcopy

Return a deep-copied publisher instance.
subscribe

Subscribe to general publisher updates.
subscribe_attr

Subscribe to updates for one named attribute.
unsubscribe_all

Remove every registered subscriber.
Source code in src/omnipy/shared/protocols/util.py 
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
class IsDataPublisher(Protocol):
    """Protocol for objects that publish change notifications to subscribers."""
    def subscribe_attr(self, attr_name: str, callback_fun: Callable[..., None]):
        """Subscribe to updates for one named attribute.

        Args:
            attr_name: Name of the published attribute to observe.
            callback_fun: Callback invoked when that attribute changes.
        """
        ...

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

        Args:
            callback_fun: Callback invoked when the publisher changes.
            do_callback: Whether to invoke the callback immediately after subscribing.
        """
        ...

    def unsubscribe_all(self) -> None:
        """Remove every registered subscriber."""
        ...

    def deepcopy(self) -> Self:
        """Return a deep-copied publisher instance.

        Returns:
            Self: Deep copy of the current publisher object.
        """
        ...

deepcopy 

deepcopy() -> Self

Return a deep-copied publisher instance.

RETURNS DESCRIPTION
Self

Deep copy of the current publisher object.

TYPE: Self

Source code in src/omnipy/shared/protocols/util.py 
32
33
34
35
36
37
38
def deepcopy(self) -> Self:
    """Return a deep-copied publisher instance.

    Returns:
        Self: Deep copy of the current publisher object.
    """
    ...

subscribe 

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

Subscribe to general publisher updates.

PARAMETER DESCRIPTION
callback_fun

Callback invoked when the publisher changes.

TYPE: Callable[..., None]

do_callback

Whether to invoke the callback immediately after subscribing.

TYPE: bool DEFAULT: True

Source code in src/omnipy/shared/protocols/util.py 
19
20
21
22
23
24
25
26
def subscribe(self, callback_fun: Callable[..., None], do_callback: bool = True) -> None:
    """Subscribe to general publisher updates.

    Args:
        callback_fun: Callback invoked when the publisher changes.
        do_callback: Whether to invoke the callback immediately after subscribing.
    """
    ...

subscribe_attr 

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

Subscribe to updates for one named attribute.

PARAMETER DESCRIPTION
attr_name

Name of the published attribute to observe.

TYPE: str

callback_fun

Callback invoked when that attribute changes.

TYPE: Callable[..., None]

Source code in src/omnipy/shared/protocols/util.py 
10
11
12
13
14
15
16
17
def subscribe_attr(self, attr_name: str, callback_fun: Callable[..., None]):
    """Subscribe to updates for one named attribute.

    Args:
        attr_name: Name of the published attribute to observe.
        callback_fun: Callback invoked when that attribute changes.
    """
    ...

unsubscribe_all 

unsubscribe_all() -> None

Remove every registered subscriber.

Source code in src/omnipy/shared/protocols/util.py 
28
29
30
def unsubscribe_all(self) -> None:
    """Remove every registered subscriber."""
    ...

IsDataclass

Bases: Protocol


              flowchart BT
              omnipy.shared.protocols.util.IsDataclass[IsDataclass]

              

              click omnipy.shared.protocols.util.IsDataclass href "" "omnipy.shared.protocols.util.IsDataclass"

Protocol for objects that expose the standard dataclass marker fields.

Source code in src/omnipy/shared/protocols/util.py 
41
42
43
44
45
46
class IsDataclass(Protocol):
    """Protocol for objects that expose the standard dataclass marker fields."""

    # as already noted in comments, checking for this attribute is currently
    # the most reliable way to ascertain that something is a dataclass
    __dataclass_fields__: ClassVar[dict[str, Any]]