Skip to content

hololinked.client.abstractions.ConsumedThingAction

Source code in hololinked\client\abstractions.py 
 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
class ConsumedThingAction:

    __slots__ = ['_resource', '_schema_validator', '__name__', '__qualname__', '__doc__']

    # action call abstraction
    # Dont add doc otherwise __doc__ in slots will conflict with class variable

    def __init__(self, 
                resource: ActionAffordance, 
                # schema_validator: typing.Type[BaseSchemaValidator] | None = None
                **kwargs
            ) -> None:
        """
        Parameters
        ----------
        resource: ActionAffordance
            dataclass object representing the action
        """
        self._resource = resource
        self._schema_validator = None # schema_validator

    def get_last_return_value(self, raise_exception: bool = False) -> typing.Any:
        """retrieve return value of the last call to the action"""
        raise NotImplementedError("implement get_last_return_value per protocol")

    last_return_value = property(fget=get_last_return_value,
                                doc="cached return value of the last call to the method")

    def __call__(self, *args, **kwargs) -> typing.Any:
        """
        Invoke action/method on server

        Parameters
        ----------
        *args: typing.Any
            arguments to the action
        **kwargs: typing.Any
            keyword arguments to the action
        """
        raise NotImplementedError("implement action _call__ per protocol")

    async def async_call(self, *args, **kwargs) -> typing.Any:
        """
        async invoke action on server - asynchronous at the network level, may not necessarily be at the server level.

        Parameters
        ----------
        *args: typing.Any
            arguments to the action
        **kwargs: typing.Any
            keyword arguments to the action
        """
        raise NotImplementedError("implement action async_call per protocol")

    def oneway(self, *args, **kwargs) -> None:
        """
        Only invokes the action on the server and does not wait for reply,
        neither does the server reply to this invokation.  

        Parameters
        ----------
        *args: typing.Any
            arguments to the action
        **kwargs: typing.Any
            keyword arguments to the action
        """
        raise NotImplementedError("implement action oneway call per protocol")

    def noblock(self, *args, **kwargs) -> str:
        """
        Invoke the action and collect the reply later

        Parameters
        ----------
        *args: typing.Any
            arguments to the action
        **kwargs: typing.Any
            keyword arguments to the action

        Returns
        -------
        str
            id of the request or message (UUID4 as string)
        """
        raise NotImplementedError("implement action noblock call per protocol")

    def __hash__(self):
        return hash(self._resource.name)

    def __eq__(self, other):
        if not isinstance(other, ConsumedThingAction):
            return False
        return self._resource.name == other._resource.name

Attributes

last_return_value class-attribute instance-attribute 

last_return_value = property(fget=get_last_return_value, doc='cached return value of the last call to the method')

Functions

__init__ 

__init__(resource: ActionAffordance, **kwargs) -> None

Parameters:

Name Type Description Default

resource

ActionAffordance

dataclass object representing the action

 required
Source code in hololinked\client\abstractions.py 
38
39
40
41
42
43
44
45
46
47
48
49
50
def __init__(self, 
            resource: ActionAffordance, 
            # schema_validator: typing.Type[BaseSchemaValidator] | None = None
            **kwargs
        ) -> None:
    """
    Parameters
    ----------
    resource: ActionAffordance
        dataclass object representing the action
    """
    self._resource = resource
    self._schema_validator = None # schema_validator

__call__ 

__call__(*args, **kwargs) -> typing.Any

Invoke action/method on server

Parameters:

Name Type Description Default

*args

arguments to the action

 ()

**kwargs

keyword arguments to the action

 {}
Source code in hololinked\client\abstractions.py 
59
60
61
62
63
64
65
66
67
68
69
70
def __call__(self, *args, **kwargs) -> typing.Any:
    """
    Invoke action/method on server

    Parameters
    ----------
    *args: typing.Any
        arguments to the action
    **kwargs: typing.Any
        keyword arguments to the action
    """
    raise NotImplementedError("implement action _call__ per protocol")

async_call async 

async_call(*args, **kwargs) -> typing.Any

async invoke action on server - asynchronous at the network level, may not necessarily be at the server level.

Parameters:

Name Type Description Default

*args

arguments to the action

 ()

**kwargs

keyword arguments to the action

 {}
Source code in hololinked\client\abstractions.py 
72
73
74
75
76
77
78
79
80
81
82
83
async def async_call(self, *args, **kwargs) -> typing.Any:
    """
    async invoke action on server - asynchronous at the network level, may not necessarily be at the server level.

    Parameters
    ----------
    *args: typing.Any
        arguments to the action
    **kwargs: typing.Any
        keyword arguments to the action
    """
    raise NotImplementedError("implement action async_call per protocol")

oneway 

oneway(*args, **kwargs) -> None

Only invokes the action on the server and does not wait for reply, neither does the server reply to this invokation.

Parameters:

Name Type Description Default

*args

arguments to the action

 ()

**kwargs

keyword arguments to the action

 {}
Source code in hololinked\client\abstractions.py 
85
86
87
88
89
90
91
92
93
94
95
96
97
def oneway(self, *args, **kwargs) -> None:
    """
    Only invokes the action on the server and does not wait for reply,
    neither does the server reply to this invokation.  

    Parameters
    ----------
    *args: typing.Any
        arguments to the action
    **kwargs: typing.Any
        keyword arguments to the action
    """
    raise NotImplementedError("implement action oneway call per protocol")

noblock 

noblock(*args, **kwargs) -> str

Invoke the action and collect the reply later

Parameters:

Name Type Description Default

*args

arguments to the action

 ()

**kwargs

keyword arguments to the action

 {}

Returns:

Type Description
str

id of the request or message (UUID4 as string)
Source code in hololinked\client\abstractions.py 
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
def noblock(self, *args, **kwargs) -> str:
    """
    Invoke the action and collect the reply later

    Parameters
    ----------
    *args: typing.Any
        arguments to the action
    **kwargs: typing.Any
        keyword arguments to the action

    Returns
    -------
    str
        id of the request or message (UUID4 as string)
    """
    raise NotImplementedError("implement action noblock call per protocol")

get_last_return_value 

get_last_return_value(raise_exception: bool = False) -> typing.Any

retrieve return value of the last call to the action

Source code in hololinked\client\abstractions.py 
52
53
54
def get_last_return_value(self, raise_exception: bool = False) -> typing.Any:
    """retrieve return value of the last call to the action"""
    raise NotImplementedError("implement get_last_return_value per protocol")