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

Data Types API (`mvi.data_types`)¶

class ArrayInput¶

Bases: numpy.ndarray

Pydantic compatible data type for numpy ndarray input.

classmethod validate(value: List) → numpy.ndarray¶

class ArrayOutput¶

Bases: numpy.ndarray

Pydantic compatible data type for numpy ndarray output.

classmethod validate(value: numpy.ndarray) → List¶

class DataFrameInput(data=None, index: Optional[Axes] = None, columns: Optional[Axes] = None, dtype: Optional[Dtype] = None, copy: bool = False)¶

Bases: pandas.core.frame.DataFrame

Pydantic compatible data type for pandas DataFrame input.

classmethod validate(value: Dict[str, Any]) → pandas.core.frame.DataFrame¶

class DataFrameOutput(data=None, index: Optional[Axes] = None, columns: Optional[Axes] = None, dtype: Optional[Dtype] = None, copy: bool = False)¶

Bases: pandas.core.frame.DataFrame

Pydantic compatible data type for pandas DataFrame output.

classmethod validate(value: pandas.core.frame.DataFrame) → Dict[str, Any]¶

Software Development Kit¶

Service API (mvi.service)¶

Communication API (mvi.communication)¶

Utilities API (mvi.utilities)¶

Data Types API (mvi.data_types)¶

Service API (`mvi.service`)¶

Communication API (`mvi.communication`)¶

Utilities API (`mvi.utilities`)¶

Data Types API (`mvi.data_types`)¶