Skip to content

Application Services API

The web application services API provides services to application services and an instance of this API is passed to the application service at construction time

WebAppServices

Source code in narvi/services/webapp_services.py 
 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
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
class WebAppServices:

    def __init__(self, webapp:"narvi.core.webapp.WebApp", workspace:str, service_id:str):
        self.app = webapp
        self.workspace = workspace
        self.service_id = service_id

    def create_request_service(self, app_name:str, handler_pattern:str,
                               handler: Callable[[dict[str, str],dict[str, str],dict[str, str], bytes], tuple[int, bytes, str, dict[str, str]]],
                               request_method:str="GET", session_id:str=None) -> tuple[str,str]:
        """
        Create a request service and return a URL to it

        Args:
            app_name: the name of the application associated with this service
            handler_pattern: specify a sub-path for the service, for example a/$b/c where components starting with $ will be resolved to request parameters
            handler: A function called when a user requests the service.  The callback is passed path parameters, request parameters, a dictionary containing HTTP headers from the request and the request body
                and should return a 4-tuple with components (response_code, binary_content, mime_type, response-headers)
            request_method: the method eg "GET", "PUT"
            session_id: the session id, if known
        Returns:
            a (handler-id, URL) that points to the created service endpoint.  The URL is relative to the app URL.
        """
        return self.app.create_request_service(app_name, handler_pattern, request_method, handler, session_id)

    def remove_request_service(self, handler_id:str):
        """
        Remove a request handler that was created by calling create_request_service

        Args:
            handler_id: the handler id returned from create_request_service when the service was created.
        """
        self.app.remove_request_service(handler_id)

    def send(self, msg:Union[str,bytes], for_session_id:str=None, except_session_id:str=None):
        """
        Send a message to the client peer (javascript) instance running in one or more client sessions

        Args:
            msg: str or bytes to send
            for_session_id: if specified, send to only this session
            except_session_id: if specified, send to all sessions except this one

        Notes:
            If both for_session_id and except_session_id are provided, except_session_id is ignored
        """
        self.app.send_webapp_message(msg, for_session_id=for_session_id, except_session_id=except_session_id)

    def add_message_listener(self, callback:Callable[[Union[str,bytes],str],None]):
        """
        Register a callback which is invoked when a message from a client peer (javascript) instance is received

        Args:
            callback: a function to call when a message is received, passing in the message content and originating session id
        """
        self.app.add_message_listener(callback)

    def add_session_open_listener(self, callback:Callable[[str,str,dict[str,str],dict[str,str]],None]):
        """
        Register a callback which is invoked when a session is opened

        Args:
            callback: a function to call when the event occurs, passing in the application name, session id, query parameters and the request headers
        """
        self.app.add_session_open_listener(callback)

    def add_session_close_listener(self, callback:Callable[[str],None]):
        """
        Register a callback which is invoked when a session is closed

        Args:
            callback: a function to call when the event occurs, passing in the session id
        """
        self.app.add_session_close_listener(callback)

    def set_metrics_callback(self, callback:Callable[[],dict[str,Any]], metrics_metadata:dict[str,dict[str,Any]]):
        """
        Register a callback which is invoked to obtain performance metrics for this instance.  This callback will be periodically
        called from a monitoring thread.

        Args:
            callback: a function to call, accepting no arguments, and returning a JSON-serialisable object mapping from metric name to value
            metrics_metadata: a dict mapping from metric name to a dict with keys "min", "max", "colour" which describe that metric
        """
        self.app.set_metrics_callback(callback, metrics_metadata)

    def set_app2app_message_callback(self, callback:Callable[[str,str,str,Any],None]):
        """
        Register a callback which is invoked when a message from another app is received

        Args:
            callback: function to be invoked when a message from another app is forwarded
                      parameters are, sending workspace_id, sending app_name, sending service_id, message data
        """
        self.app.set_app2app_message_callback(callback)

    def set_admin_listener(self, callback:Callable[[dict[str,Any]],None]):
        """
        Register a callback which is invoked with the latest admin information on all apps/services, including metrics

        Args:
            callback: a function to call, accepting no arguments, and returning a JSON-serialisable object
        """
        self.app.set_admin_listener(callback)

    def add_app_close_listener(self, callback:Callable[[],None]):
        """
        Register a callback which is invoked when the application instance is about to be closed

        Args:
            callback: a function to call when the event occurs, accepting no arguments
        """
        self.app.add_app_close_listener(callback)

    def get_service_id(self) -> str:
        """
        Gets the service id associated with this instance of the app service

        Returns:
            string containing the service id
        """
        return self.service_id

    def get_workspace_id(self) -> str:
        """
        Gets the workspace id in which this instance of the app service is running

        Returns:
            string containing the workspace id
        """
        return self.workspace

    def get_app_name(self) -> str:
        """
        Gets the application service name of this application service

        Returns:
            string containing the app name
        """
        return self.app.get_name()

    def send_to_app(self, workspace_id:str, app_name:str, service_id:str, data:Any) -> bool:
        """
        Send a message to another app

        Args:
            workspace_id: the workspace id of the app
            app_name: the application name
            service_id: the service id
            data: the data to send in the message

        Returns:
            True iff the receiving service instance was located to receive the message, False otherwise
        """
        return self.app.send_to_app(to_workspace_id=workspace_id, to_app_name=app_name, to_service_id=service_id, data=data)

add_app_close_listener(callback)

Register a callback which is invoked when the application instance is about to be closed

Parameters:

Name Type Description Default
callback Callable[[], None]

a function to call when the event occurs, accepting no arguments

 required
Source code in narvi/services/webapp_services.py 
137
138
139
140
141
142
143
144
def add_app_close_listener(self, callback:Callable[[],None]):
    """
    Register a callback which is invoked when the application instance is about to be closed

    Args:
        callback: a function to call when the event occurs, accepting no arguments
    """
    self.app.add_app_close_listener(callback)

add_message_listener(callback)

Register a callback which is invoked when a message from a client peer (javascript) instance is received

Parameters:

Name Type Description Default
callback Callable[[Union[str, bytes], str], None]

a function to call when a message is received, passing in the message content and originating session id

 required
Source code in narvi/services/webapp_services.py 
80
81
82
83
84
85
86
87
def add_message_listener(self, callback:Callable[[Union[str,bytes],str],None]):
    """
    Register a callback which is invoked when a message from a client peer (javascript) instance is received

    Args:
        callback: a function to call when a message is received, passing in the message content and originating session id
    """
    self.app.add_message_listener(callback)

add_session_close_listener(callback)

Register a callback which is invoked when a session is closed

Parameters:

Name Type Description Default
callback Callable[[str], None]

a function to call when the event occurs, passing in the session id

 required
Source code in narvi/services/webapp_services.py 
 98
 99
100
101
102
103
104
105
def add_session_close_listener(self, callback:Callable[[str],None]):
    """
    Register a callback which is invoked when a session is closed

    Args:
        callback: a function to call when the event occurs, passing in the session id
    """
    self.app.add_session_close_listener(callback)

add_session_open_listener(callback)

Register a callback which is invoked when a session is opened

Parameters:

Name Type Description Default
callback Callable[[str, str, dict[str, str], dict[str, str]], None]

a function to call when the event occurs, passing in the application name, session id, query parameters and the request headers

 required
Source code in narvi/services/webapp_services.py 
89
90
91
92
93
94
95
96
def add_session_open_listener(self, callback:Callable[[str,str,dict[str,str],dict[str,str]],None]):
    """
    Register a callback which is invoked when a session is opened

    Args:
        callback: a function to call when the event occurs, passing in the application name, session id, query parameters and the request headers
    """
    self.app.add_session_open_listener(callback)

create_request_service(app_name, handler_pattern, handler, request_method='GET', session_id=None)

Create a request service and return a URL to it

Parameters:

Name Type Description Default
app_name str

the name of the application associated with this service

 required
handler_pattern str

specify a sub-path for the service, for example a/$b/c where components starting with $ will be resolved to request parameters

 required
handler Callable[[dict[str, str], dict[str, str], dict[str, str], bytes], tuple[int, bytes, str, dict[str, str]]]

A function called when a user requests the service. The callback is passed path parameters, request parameters, a dictionary containing HTTP headers from the request and the request body and should return a 4-tuple with components (response_code, binary_content, mime_type, response-headers)

 required
request_method str

the method eg "GET", "PUT"

 'GET'
session_id str

the session id, if known

 None

Returns: a (handler-id, URL) that points to the created service endpoint. The URL is relative to the app URL.

Source code in narvi/services/webapp_services.py 
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
def create_request_service(self, app_name:str, handler_pattern:str,
                           handler: Callable[[dict[str, str],dict[str, str],dict[str, str], bytes], tuple[int, bytes, str, dict[str, str]]],
                           request_method:str="GET", session_id:str=None) -> tuple[str,str]:
    """
    Create a request service and return a URL to it

    Args:
        app_name: the name of the application associated with this service
        handler_pattern: specify a sub-path for the service, for example a/$b/c where components starting with $ will be resolved to request parameters
        handler: A function called when a user requests the service.  The callback is passed path parameters, request parameters, a dictionary containing HTTP headers from the request and the request body
            and should return a 4-tuple with components (response_code, binary_content, mime_type, response-headers)
        request_method: the method eg "GET", "PUT"
        session_id: the session id, if known
    Returns:
        a (handler-id, URL) that points to the created service endpoint.  The URL is relative to the app URL.
    """
    return self.app.create_request_service(app_name, handler_pattern, request_method, handler, session_id)

get_app_name()

Gets the application service name of this application service

Returns:

Type Description
str

string containing the app name
Source code in narvi/services/webapp_services.py 
164
165
166
167
168
169
170
171
def get_app_name(self) -> str:
    """
    Gets the application service name of this application service

    Returns:
        string containing the app name
    """
    return self.app.get_name()

get_service_id()

Gets the service id associated with this instance of the app service

Returns:

Type Description
str

string containing the service id
Source code in narvi/services/webapp_services.py 
146
147
148
149
150
151
152
153
def get_service_id(self) -> str:
    """
    Gets the service id associated with this instance of the app service

    Returns:
        string containing the service id
    """
    return self.service_id

get_workspace_id()

Gets the workspace id in which this instance of the app service is running

Returns:

Type Description
str

string containing the workspace id
Source code in narvi/services/webapp_services.py 
155
156
157
158
159
160
161
162
def get_workspace_id(self) -> str:
    """
    Gets the workspace id in which this instance of the app service is running

    Returns:
        string containing the workspace id
    """
    return self.workspace

remove_request_service(handler_id)

Remove a request handler that was created by calling create_request_service

Parameters:

Name Type Description Default
handler_id str

the handler id returned from create_request_service when the service was created.

 required
Source code in narvi/services/webapp_services.py 
57
58
59
60
61
62
63
64
def remove_request_service(self, handler_id:str):
    """
    Remove a request handler that was created by calling create_request_service

    Args:
        handler_id: the handler id returned from create_request_service when the service was created.
    """
    self.app.remove_request_service(handler_id)

send(msg, for_session_id=None, except_session_id=None)

Send a message to the client peer (javascript) instance running in one or more client sessions

Parameters:

Name Type Description Default
msg Union[str, bytes]

str or bytes to send

 required
for_session_id str

if specified, send to only this session

 None
except_session_id str

if specified, send to all sessions except this one

 None
Notes

If both for_session_id and except_session_id are provided, except_session_id is ignored

Source code in narvi/services/webapp_services.py 
66
67
68
69
70
71
72
73
74
75
76
77
78
def send(self, msg:Union[str,bytes], for_session_id:str=None, except_session_id:str=None):
    """
    Send a message to the client peer (javascript) instance running in one or more client sessions

    Args:
        msg: str or bytes to send
        for_session_id: if specified, send to only this session
        except_session_id: if specified, send to all sessions except this one

    Notes:
        If both for_session_id and except_session_id are provided, except_session_id is ignored
    """
    self.app.send_webapp_message(msg, for_session_id=for_session_id, except_session_id=except_session_id)

send_to_app(workspace_id, app_name, service_id, data)

Send a message to another app

Parameters:

Name Type Description Default
workspace_id str

the workspace id of the app

 required
app_name str

the application name

 required
service_id str

the service id

 required
data Any

the data to send in the message

 required

Returns:

Type Description
bool

True iff the receiving service instance was located to receive the message, False otherwise
Source code in narvi/services/webapp_services.py 
173
174
175
176
177
178
179
180
181
182
183
184
185
186
def send_to_app(self, workspace_id:str, app_name:str, service_id:str, data:Any) -> bool:
    """
    Send a message to another app

    Args:
        workspace_id: the workspace id of the app
        app_name: the application name
        service_id: the service id
        data: the data to send in the message

    Returns:
        True iff the receiving service instance was located to receive the message, False otherwise
    """
    return self.app.send_to_app(to_workspace_id=workspace_id, to_app_name=app_name, to_service_id=service_id, data=data)

set_admin_listener(callback)

Register a callback which is invoked with the latest admin information on all apps/services, including metrics

Parameters:

Name Type Description Default
callback Callable[[dict[str, Any]], None]

a function to call, accepting no arguments, and returning a JSON-serialisable object

 required
Source code in narvi/services/webapp_services.py 
128
129
130
131
132
133
134
135
def set_admin_listener(self, callback:Callable[[dict[str,Any]],None]):
    """
    Register a callback which is invoked with the latest admin information on all apps/services, including metrics

    Args:
        callback: a function to call, accepting no arguments, and returning a JSON-serialisable object
    """
    self.app.set_admin_listener(callback)

set_app2app_message_callback(callback)

Register a callback which is invoked when a message from another app is received

Parameters:

Name Type Description Default
callback Callable[[str, str, str, Any], None]

function to be invoked when a message from another app is forwarded parameters are, sending workspace_id, sending app_name, sending service_id, message data

 required
Source code in narvi/services/webapp_services.py 
118
119
120
121
122
123
124
125
126
def set_app2app_message_callback(self, callback:Callable[[str,str,str,Any],None]):
    """
    Register a callback which is invoked when a message from another app is received

    Args:
        callback: function to be invoked when a message from another app is forwarded
                  parameters are, sending workspace_id, sending app_name, sending service_id, message data
    """
    self.app.set_app2app_message_callback(callback)

set_metrics_callback(callback, metrics_metadata)

Register a callback which is invoked to obtain performance metrics for this instance. This callback will be periodically called from a monitoring thread.

Parameters:

Name Type Description Default
callback Callable[[], dict[str, Any]]

a function to call, accepting no arguments, and returning a JSON-serialisable object mapping from metric name to value

 required
metrics_metadata dict[str, dict[str, Any]]

a dict mapping from metric name to a dict with keys "min", "max", "colour" which describe that metric

 required
Source code in narvi/services/webapp_services.py 
107
108
109
110
111
112
113
114
115
116
def set_metrics_callback(self, callback:Callable[[],dict[str,Any]], metrics_metadata:dict[str,dict[str,Any]]):
    """
    Register a callback which is invoked to obtain performance metrics for this instance.  This callback will be periodically
    called from a monitoring thread.

    Args:
        callback: a function to call, accepting no arguments, and returning a JSON-serialisable object mapping from metric name to value
        metrics_metadata: a dict mapping from metric name to a dict with keys "min", "max", "colour" which describe that metric
    """
    self.app.set_metrics_callback(callback, metrics_metadata)