Software Development Kit

Service API (daeploy.service)

Note

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

daeploy.service
entrypoint(self, func: Optional[Callable] = None, method: str = 'POST', monitor: bool = False, disable_http_logs: 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.

  • method (str) – HTTP method for entrypoint. Defauts to “POST”

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

  • disable_http_logs (bool) – Set if the http entry logs should be disabled for this entrypoint. These logs are genereated from uvicorn. Defaults to False. Example of http entry log: "POST /services/service_1.0.0/entrypoint_name HTTP/1.1" 200 OK

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

Raises
  • TypeError – If func is not callable.

  • ValueError – If method is not a valid HTTP method.

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 JSON with json.dumps()

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 parameter value

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, disable_http_logs: 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.

  • disable_http_logs (bool) – Disable logs when getting and setting paramter with API. 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 (daeploy.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, entrypoint_method: str = 'POST', **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, …}. Defaults to None.

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

  • entrypoint_method (str) – HTTP method of the entrypoint to call. You only need to change this if you have created an entrypoint with a non-default HTTP method. Defaults to “POST”.

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

Raises

ValueError – If entrypoint_method is not a valid HTTP method

Returns

The output from the entrypoint in the other service.

Return type

Any

notify(msg: str, severity: daeploy.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 daeploy 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 (daeploy.utilities)

get_authorized_domains() list

Returns list of autorized domains for auth header

Returns

List of authorized domains

Return type

list

get_daeploy_manager_hostname() str

Returns the hostname of the manager currently running this service.

Returns

Hostname of manager

Return type

str

get_daeploy_manager_url() str

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

Returns

URL to manager

Return type

str

get_db_clean_interval_seconds() float

Converts the environment variable DAEPLOY_SERVICE_DB_CLEAN_INTERVAL, that defines how often to clean the database from old records, into seconds and returns that.

Returns

Seconds between cleans. Defaults to 7 days

Return type

float

get_db_table_limit() Tuple[int]

Limit of rows or duration to keep records in service database. Reads from the environment variable DAEPLOY_SERVICE_DB_TABLE_LIMIT. Data will be cleaned in intervals, defined by the environment variable DAEPLOY_SERVICE_DB_CLEAN_INTERVAL.

Returns

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

  • 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_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

match_limit_and_unit(string: str, accepted_units: List[str]) Tuple[str]

Data Types API (daeploy.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[Collection[Any]] = None, columns: Optional[Collection[Any]] = None, dtype: Optional[Union[pandas.core.dtypes.base.ExtensionDtype, str, numpy.dtype, Type[Union[str, float, int, complex, bool, object]]]] = None, copy: Optional[bool] = None)

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[Collection[Any]] = None, columns: Optional[Collection[Any]] = None, dtype: Optional[Union[pandas.core.dtypes.base.ExtensionDtype, str, numpy.dtype, Type[Union[str, float, int, complex, bool, object]]]] = None, copy: Optional[bool] = None)

Bases: pandas.core.frame.DataFrame

Pydantic compatible data type for pandas DataFrame output.

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