Software Development Kit

Service API (mvi.service)

Note

mvi.service is an object (not a module), and should be used accordingly.

mvi.service

entrypoint(self, func: Optional[Callable] = None, monitor: bool = False, **fastapi_kwargs) → Callable

Registers a function as an entrypoint, which will make it reachable as an HTTP method on your host machine.

Decorate a function with this method to create an entrypoint for it:

@entrypoint
def my_function(arg1:type1) -> type2:
    ....

It is strongly recommended to include types of the arguments and return objects for the decorated function.

Parameters

• func (Callable) – The decorated function to make an entrypoint for.

• monitor (bool) – Set if the input and output to this entrypoint should be saved to the service’s monitoring database.

• **fastapi_kwargs – Keyword arguments for the resulting API endpoint. See FastAPI for keyword arguments of the FastAPI.post() function.

Raises

TypeError – If func is not callable.

Returns

The decorated function: func.

Return type

Callable

store(self, **variables)

Saves variables to the service’s monitoring database. Supports numbers and strings. If a variable is not a number or string it store will try to coerce it into a string before storing.

Parameters

**variables – Variables to save to the database. Non-numeric variables will be saved as str(variable).

call_every(self, seconds: float, wait_first: bool = False)

Returns a decorator that converts a function to an awaitable that runs every seconds.

Decorate a function with this method to make it run repeatedly:

@call_every(seconds=60)
def my_function():
    ....

Parameters

• seconds (float) – Interval between calls in seconds

• wait_first (bool) – If we should skip the first execution. Defaults to False.

Returns

The decorator

Return type

Callable

get_parameter(self, parameter: str) → Any

Get specific parameter

Parameters

parameter (str) – The name of the parameter

Returns

The value of the parameter

Return type

Any

add_parameter(self, parameter: str, value: Any, expose: bool = True, monitor: bool = False)

Adds a parameter to the parameter endpoints.

Parameters

• parameter (str) – The name of the parameter

• value (Any) – The value of the parameter

• expose (bool) – Should parameter update be exposed to the API. Defaults to True.

• monitor (bool) – Stores updates to this parameter in the monitoring database if True. Will try to coerce non-numeric types to string Defaults to False.

run(self)

Runs the service

This method is usually called at the end of the module when all entrypoints etc for the service has been specified

Communication API (mvi.communication)

class Severity(value)

Bases: enum.Enum

An enumeration of notification severities.

CRITICAL = 2

INFO = 0

WARNING = 1

call_service(service_name: str, entrypoint_name: str, arguments: Optional[dict] = None, service_version: Optional[str] = None, **request_kwargs) → Any

Call an entrypoint in a different service.

Example usage:

data = call_service(
    service_name="data_connector",
    entrypoint_name= "get_data",
    arguments={"variable_name": value}
)

Parameters

• service_name (str) – The service which contains the entrypoint to call.

• entrypoint_name (str) – The name of the entrypoint to call. The return object(s) of this entrypoint must be jsonable, i.e pass FastAPI’s jsonable_encoder, otherwise it won’t be reachable.

• arguments (dict) – Arguments to the entrypoint. In the form: {“argument_name”: value, …}. Default None.

• service_version (str) – The specific version of the service to call. Default None, the main version and the shadows versions will be called.

• **request_kwargs – Keyword arguments to pass on to :func:requests.post.

Returns

The output from the entrypoint in the other service.

Return type

Any

notify(msg: str, severity: mvi.communication.Severity, dashboard: bool = True, emails: Optional[List[str]] = None, timer: int = 0)

Generates and sends a notification

Parameters

• msg (str) – The message to include for the recipient of the notification.

• severity (Severity) – The severity of the notification.

• dashboard (bool) – If this is True, the notification will be shown on the mvi dashboard. Defaults to True.

• emails (List[str]) – List of emails to send the notifications to. Defaults to None, in which case no emails are sent

• timer (int) – The amount of time (in seconds) that has to pass before the same notification can be send again. Defaults to 0.

Utilities API (mvi.utilities)

get_authorized_domains() → list

Returns list of autorized domains for auth header

Returns

List of authorized domains

Return type

list

get_db_table_limit() → int

Limit of rows or duration to keep records in service database. Reads from the environment variable MVI_SERVICE_DB_TABLE_LIMIT.

Returns

• int: Maximum rows/days etc. of database. Defaults to 365

• str: One of rows, days, hours, minutes, seconds. Defaults to “days”

Return type

tuple

get_headers() → dict

Generating headers for API calls

Returns

Assembled header

Return type

dict

get_mvi_manager_hostname() → str

Returns the hostname of the manager currently running this service.

Returns

Hostname of manager

Return type

str

get_mvi_manager_url() → str

Returns a URL where the manager currently running this service can be reached.

Returns

URL to manager

Return type

str

get_service_access_token() → str

Returns a valid access token for communicating with the manager currently running this service and other services running on that manager.

Returns

JWT token

Return type

str

get_service_name() → str

Name of this service

Returns

Name

Return type

str

get_service_root_path() → str

Returns the root path at which this service is running. Can be for example /services/<service_name>_<service_version>

Returns

Root path

Return type

str

get_service_version() → str

Versio of this service

Returns

Version string (ex. 1.2.3)

Return type

str