Skip to content

Server API

The server API allows the caller to create and configure a narvi server.

NarviServer

Source code in narvi/api/narvi_server.py 
 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
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
class NarviServer:

    def __init__(self, host: str = "localhost", port: int = 8999, web_server_type: str = "tornado",
                 base_path: str = "/narvi", admin_path:str=None, monitoring_interval_s:int=None,
                 monitoring_retention_s:int=None):
        """
        Constructs a Narvi app server

        Args:
            host: hostname to run server, for example "localhost" or "0.0.0.0"
            port: port number for the servier, for example, 8080
            web_server_type: either "tornado" (recommended) or "builtin"
            base_path: the base url for all narvi URLs, should start with a leading "/"
            admin_path: path to listen at for admin requests
            monitoring_interval_s: the interval in seconds to poll webapps for metrics, or None to turn off monitoring
            monitoring_retention_s: discard metrics collected more than this many seconds ago, or None to never discard
        """
        self.service = Service(host=host, port=port, base_path=base_path,
                               web_server_type=web_server_type,
                               admin_path=admin_path,
                               monitoring_interval=monitoring_interval_s,
                               monitoring_retention=monitoring_retention_s)
        self.logger = logging.getLogger("NaviServer")

    def register_service(self, workspace:str, app_service_name:str, app_cls_name:str, app_parameters:dict[str,Any]={},
                         fixed_service_id:str=None, shared_service:bool=True, service_id_validator:Callable[[str],bool]=lambda service_id:True,
                         idle_timeout:int=3600) -> RegisteredService:
        """
        Register an application service backend.  Websocket connections can be made to this service at URL

        base_path + /$workspace/$app_service_name/$service_id/connect

        Args:
            workspace: a string identifying a workspace within which the app will be registered.  The workspace may be used as a security context.
            app_service_name: the name of the app
            app_cls_name: the resource path and name of the class implementing the web app name, eg foo.bar.foobar.FooBar
            app_parameters: a set of parameters passed to the app constructor
            fixed_service_id:  set this parameter if the service instances will be associated with this service id only
            shared_service: whether the service allows multiple connections to the same service instance
            service_id_validator: a callable which validates the service id
            idle_timeout: for shared services with no fixed service id - close the service after this delay once the last client disconnects

        Returns:
            a RegisteredService instance
        """
        return self.service.register_service(workspace=workspace, app_service_name=app_service_name, app_cls_name=app_cls_name,
                                             app_parameters=app_parameters, fixed_service_id=fixed_service_id,
                                             shared_service=shared_service, service_id_validator=service_id_validator,
                                             idle_timeout=idle_timeout)

    def register_app(self, app_name:str, application_service:RegisteredService, app_parameters:dict[str,JsonType]={},
                     resource_roots:dict[str|tuple[str,str],str]={}, service_chooser_app_name:str=None):
        """
        Register a web application frontend.  The application can be loaded from the following URL

        base_path + /$workspace/$app_service_name/$service_id/index.html
            or
        base_path + /$workspace/$app_service_name/index.html

        Args:
            app_name: the name of the app
            application_service: the name of the backend application service this frontend will connect to
            app_parameters: parameters to pass to the web application when constructed
            resource_roots: a dictionary mapping from a URL (relative to the application URL) to a filesystem path
                            eg individual files: { "images/icon.png": "/full/path/to/images/icon.png" }
                            or folders: { ("images","*.png"): "/full/path/to/images" }
                            both of these definitions will load URL images/icon.png from /full/path/to/images/icon.png
                            but the second definition will load all png images files from that folder
            service_chooser_app_name: the name of an app to redirect to if the service id is not provided by a connecting client
        """

        self.service.register_app(app_name, application_service=application_service,
                                  app_parameters=app_parameters,
                                  resource_roots=resource_roots,
                                  service_chooser_app_name=service_chooser_app_name)

    def register_redirect(self, from_url:str, to_workspace:str, to_app:str):
        """
        Create a redirection from the given URL to a given workspace and app

        Args:
            from_url: the URL to redirect, for example foo/bar
            to_workspace: the target workspace
            to_app: the target application name
        """
        self.service.register_redirect(from_url, to_workspace, to_app)

    def list_app_urls(self) -> list[tuple[str,str,str]]:
        """
        Find out which services are available

        Returns:
            a list of tuples (workspace, app_name, url) describing the apps that are registered
        """
        return self.service.get_summary()

    def run(self, ready_callback:Callable[[],None]=None):
        """
        Run the server

        Args:
            ready_callback: optional, a function that will be called when the server is listening and ready to accept connections
        """
        self.service.run(ready_callback)

    def start_service(self, workspace:str, app_service_name:str, service_id:str):
        """
        Start an instance of a service running

        Args:
            workspace: the workspace within which the service will run
            app_service_name: the name of the application service - must match a registered application service
            service_id: the service id for the service
        """
        self.service.start_service(workspace, app_service_name, service_id)

    def restart_service(self, workspace:str, app_service_name:str, service_id:str):
        """
        Restart a service

        Args:
            workspace: the workspace within which the service is running
            app_service_name: the name of the application service
            service_id: the service id
        """
        self.service.restart_service(workspace, app_service_name, service_id)

    def stop_service(self, workspace:str, app_service_name:str, service_id:str):
        """
        Stop a service

        Args:
            workspace: the workspace within which the service is running
            app_service_name: the name of the application service
            service_id: the service id
        """
        self.service.stop_service(workspace, app_service_name, service_id)

    def get_service_statuses(self, for_workspace:str) -> dict[str,JsonType]:
        """
        Get details of all services running in the workspace

        Args:
            for_workspace: the workspace to get details for

        Returns:
            dictionary mapping from service_id to service details for running services only
        """
        return self.service.get_admin_data(for_workspace=for_workspace)

    def close(self):
        """
        Stop and close this server.
        """
        self.service.close()

    def get_path_of_resource(self, package:str, resource:str="") -> str:
        """
        Gets the filesystem path to a python package or a resource within the package

        Args:
            package: the package name, expressed using dotted notation, for example module1.submoduleA
            resource: optional, provides the name of a file within the resource to load

        Returns:
            the filesystem path to the package directory or resource file.

        Raises:
            ModuleNotFoundError: if no package could be found
        """
        return ResourceLoader.get_path_of_resource(package, resource)

__init__(host='localhost', port=8999, web_server_type='tornado', base_path='/narvi', admin_path=None, monitoring_interval_s=None, monitoring_retention_s=None)

Constructs a Narvi app server

Parameters:

Name Type Description Default
host str

hostname to run server, for example "localhost" or "0.0.0.0"

 'localhost'
port int

port number for the servier, for example, 8080

 8999
web_server_type str

either "tornado" (recommended) or "builtin"

 'tornado'
base_path str

the base url for all narvi URLs, should start with a leading "/"

 '/narvi'
admin_path str

path to listen at for admin requests

 None
monitoring_interval_s int

the interval in seconds to poll webapps for metrics, or None to turn off monitoring

 None
monitoring_retention_s int

discard metrics collected more than this many seconds ago, or None to never discard

 None
Source code in narvi/api/narvi_server.py 
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
def __init__(self, host: str = "localhost", port: int = 8999, web_server_type: str = "tornado",
             base_path: str = "/narvi", admin_path:str=None, monitoring_interval_s:int=None,
             monitoring_retention_s:int=None):
    """
    Constructs a Narvi app server

    Args:
        host: hostname to run server, for example "localhost" or "0.0.0.0"
        port: port number for the servier, for example, 8080
        web_server_type: either "tornado" (recommended) or "builtin"
        base_path: the base url for all narvi URLs, should start with a leading "/"
        admin_path: path to listen at for admin requests
        monitoring_interval_s: the interval in seconds to poll webapps for metrics, or None to turn off monitoring
        monitoring_retention_s: discard metrics collected more than this many seconds ago, or None to never discard
    """
    self.service = Service(host=host, port=port, base_path=base_path,
                           web_server_type=web_server_type,
                           admin_path=admin_path,
                           monitoring_interval=monitoring_interval_s,
                           monitoring_retention=monitoring_retention_s)
    self.logger = logging.getLogger("NaviServer")

close()

Stop and close this server.

Source code in narvi/api/narvi_server.py 
184
185
186
187
188
def close(self):
    """
    Stop and close this server.
    """
    self.service.close()

get_path_of_resource(package, resource='')

Gets the filesystem path to a python package or a resource within the package

Parameters:

Name Type Description Default
package str

the package name, expressed using dotted notation, for example module1.submoduleA

 required
resource str

optional, provides the name of a file within the resource to load

 ''

Returns:

Type Description
str

the filesystem path to the package directory or resource file.

Raises:

Type Description
ModuleNotFoundError

if no package could be found
Source code in narvi/api/narvi_server.py 
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
def get_path_of_resource(self, package:str, resource:str="") -> str:
    """
    Gets the filesystem path to a python package or a resource within the package

    Args:
        package: the package name, expressed using dotted notation, for example module1.submoduleA
        resource: optional, provides the name of a file within the resource to load

    Returns:
        the filesystem path to the package directory or resource file.

    Raises:
        ModuleNotFoundError: if no package could be found
    """
    return ResourceLoader.get_path_of_resource(package, resource)

get_service_statuses(for_workspace)

Get details of all services running in the workspace

Parameters:

Name Type Description Default
for_workspace str

the workspace to get details for

 required

Returns:

Type Description
dict[str, JsonType]

dictionary mapping from service_id to service details for running services only
Source code in narvi/api/narvi_server.py 
172
173
174
175
176
177
178
179
180
181
182
def get_service_statuses(self, for_workspace:str) -> dict[str,JsonType]:
    """
    Get details of all services running in the workspace

    Args:
        for_workspace: the workspace to get details for

    Returns:
        dictionary mapping from service_id to service details for running services only
    """
    return self.service.get_admin_data(for_workspace=for_workspace)

list_app_urls()

Find out which services are available

Returns:

Type Description
list[tuple[str, str, str]]

a list of tuples (workspace, app_name, url) describing the apps that are registered
Source code in narvi/api/narvi_server.py 
121
122
123
124
125
126
127
128
def list_app_urls(self) -> list[tuple[str,str,str]]:
    """
    Find out which services are available

    Returns:
        a list of tuples (workspace, app_name, url) describing the apps that are registered
    """
    return self.service.get_summary()

register_app(app_name, application_service, app_parameters={}, resource_roots={}, service_chooser_app_name=None)

Register a web application frontend. The application can be loaded from the following URL

base_path + /$workspace/$app_service_name/$service_id/index.html or base_path + /$workspace/$app_service_name/index.html

Parameters:

Name Type Description Default
app_name str

the name of the app

 required
application_service RegisteredService

the name of the backend application service this frontend will connect to

 required
app_parameters dict[str, JsonType]

parameters to pass to the web application when constructed

 {}
resource_roots dict[str | tuple[str, str], str]

a dictionary mapping from a URL (relative to the application URL) to a filesystem path eg individual files: { "images/icon.png": "/full/path/to/images/icon.png" } or folders: { ("images","*.png"): "/full/path/to/images" } both of these definitions will load URL images/icon.png from /full/path/to/images/icon.png but the second definition will load all png images files from that folder

 {}
service_chooser_app_name str

the name of an app to redirect to if the service id is not provided by a connecting client

 None
Source code in narvi/api/narvi_server.py 
 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
def register_app(self, app_name:str, application_service:RegisteredService, app_parameters:dict[str,JsonType]={},
                 resource_roots:dict[str|tuple[str,str],str]={}, service_chooser_app_name:str=None):
    """
    Register a web application frontend.  The application can be loaded from the following URL

    base_path + /$workspace/$app_service_name/$service_id/index.html
        or
    base_path + /$workspace/$app_service_name/index.html

    Args:
        app_name: the name of the app
        application_service: the name of the backend application service this frontend will connect to
        app_parameters: parameters to pass to the web application when constructed
        resource_roots: a dictionary mapping from a URL (relative to the application URL) to a filesystem path
                        eg individual files: { "images/icon.png": "/full/path/to/images/icon.png" }
                        or folders: { ("images","*.png"): "/full/path/to/images" }
                        both of these definitions will load URL images/icon.png from /full/path/to/images/icon.png
                        but the second definition will load all png images files from that folder
        service_chooser_app_name: the name of an app to redirect to if the service id is not provided by a connecting client
    """

    self.service.register_app(app_name, application_service=application_service,
                              app_parameters=app_parameters,
                              resource_roots=resource_roots,
                              service_chooser_app_name=service_chooser_app_name)

register_redirect(from_url, to_workspace, to_app)

Create a redirection from the given URL to a given workspace and app

Parameters:

Name Type Description Default
from_url str

the URL to redirect, for example foo/bar

 required
to_workspace str

the target workspace

 required
to_app str

the target application name

 required
Source code in narvi/api/narvi_server.py 
110
111
112
113
114
115
116
117
118
119
def register_redirect(self, from_url:str, to_workspace:str, to_app:str):
    """
    Create a redirection from the given URL to a given workspace and app

    Args:
        from_url: the URL to redirect, for example foo/bar
        to_workspace: the target workspace
        to_app: the target application name
    """
    self.service.register_redirect(from_url, to_workspace, to_app)

register_service(workspace, app_service_name, app_cls_name, app_parameters={}, fixed_service_id=None, shared_service=True, service_id_validator=lambda service_id: True, idle_timeout=3600)

Register an application service backend. Websocket connections can be made to this service at URL

base_path + /$workspace/$app_service_name/$service_id/connect

Parameters:

Name Type Description Default
workspace str

a string identifying a workspace within which the app will be registered. The workspace may be used as a security context.

 required
app_service_name str

the name of the app

 required
app_cls_name str

the resource path and name of the class implementing the web app name, eg foo.bar.foobar.FooBar

 required
app_parameters dict[str, Any]

a set of parameters passed to the app constructor

 {}
fixed_service_id str

set this parameter if the service instances will be associated with this service id only

 None
shared_service bool

whether the service allows multiple connections to the same service instance

 True
service_id_validator Callable[[str], bool]

a callable which validates the service id

 lambda service_id: True
idle_timeout int

for shared services with no fixed service id - close the service after this delay once the last client disconnects

 3600

Returns:

Type Description
RegisteredService

a RegisteredService instance
Source code in narvi/api/narvi_server.py 
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
def register_service(self, workspace:str, app_service_name:str, app_cls_name:str, app_parameters:dict[str,Any]={},
                     fixed_service_id:str=None, shared_service:bool=True, service_id_validator:Callable[[str],bool]=lambda service_id:True,
                     idle_timeout:int=3600) -> RegisteredService:
    """
    Register an application service backend.  Websocket connections can be made to this service at URL

    base_path + /$workspace/$app_service_name/$service_id/connect

    Args:
        workspace: a string identifying a workspace within which the app will be registered.  The workspace may be used as a security context.
        app_service_name: the name of the app
        app_cls_name: the resource path and name of the class implementing the web app name, eg foo.bar.foobar.FooBar
        app_parameters: a set of parameters passed to the app constructor
        fixed_service_id:  set this parameter if the service instances will be associated with this service id only
        shared_service: whether the service allows multiple connections to the same service instance
        service_id_validator: a callable which validates the service id
        idle_timeout: for shared services with no fixed service id - close the service after this delay once the last client disconnects

    Returns:
        a RegisteredService instance
    """
    return self.service.register_service(workspace=workspace, app_service_name=app_service_name, app_cls_name=app_cls_name,
                                         app_parameters=app_parameters, fixed_service_id=fixed_service_id,
                                         shared_service=shared_service, service_id_validator=service_id_validator,
                                         idle_timeout=idle_timeout)

restart_service(workspace, app_service_name, service_id)

Restart a service

Parameters:

Name Type Description Default
workspace str

the workspace within which the service is running

 required
app_service_name str

the name of the application service

 required
service_id str

the service id

 required
Source code in narvi/api/narvi_server.py 
150
151
152
153
154
155
156
157
158
159
def restart_service(self, workspace:str, app_service_name:str, service_id:str):
    """
    Restart a service

    Args:
        workspace: the workspace within which the service is running
        app_service_name: the name of the application service
        service_id: the service id
    """
    self.service.restart_service(workspace, app_service_name, service_id)

run(ready_callback=None)

Run the server

Parameters:

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

optional, a function that will be called when the server is listening and ready to accept connections

 None
Source code in narvi/api/narvi_server.py 
130
131
132
133
134
135
136
137
def run(self, ready_callback:Callable[[],None]=None):
    """
    Run the server

    Args:
        ready_callback: optional, a function that will be called when the server is listening and ready to accept connections
    """
    self.service.run(ready_callback)

start_service(workspace, app_service_name, service_id)

Start an instance of a service running

Parameters:

Name Type Description Default
workspace str

the workspace within which the service will run

 required
app_service_name str

the name of the application service - must match a registered application service

 required
service_id str

the service id for the service

 required
Source code in narvi/api/narvi_server.py 
139
140
141
142
143
144
145
146
147
148
def start_service(self, workspace:str, app_service_name:str, service_id:str):
    """
    Start an instance of a service running

    Args:
        workspace: the workspace within which the service will run
        app_service_name: the name of the application service - must match a registered application service
        service_id: the service id for the service
    """
    self.service.start_service(workspace, app_service_name, service_id)

stop_service(workspace, app_service_name, service_id)

Stop a service

Parameters:

Name Type Description Default
workspace str

the workspace within which the service is running

 required
app_service_name str

the name of the application service

 required
service_id str

the service id

 required
Source code in narvi/api/narvi_server.py 
161
162
163
164
165
166
167
168
169
170
def stop_service(self, workspace:str, app_service_name:str, service_id:str):
    """
    Stop a service

    Args:
        workspace: the workspace within which the service is running
        app_service_name: the name of the application service
        service_id: the service id
    """
    self.service.stop_service(workspace, app_service_name, service_id)