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

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: 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: ndarray

Pydantic compatible data type for numpy ndarray input.

classmethod validate(value: List) → ndarray

class ArrayOutput

Bases: ndarray

Pydantic compatible data type for numpy ndarray output.

classmethod validate(value: ndarray) → List

class DataFrameInput(data=None, index: Optional[Collection[Any]] = None, columns: Optional[Collection[Any]] = None, dtype: Optional[Union[ExtensionDtype, str, dtype, Type[Union[str, float, int, complex, bool, object]]]] = None, copy: Optional[bool] = None)

Bases: DataFrame

Pydantic compatible data type for pandas DataFrame input.

classmethod validate(value: Dict[str, Any]) → DataFrame

class DataFrameOutput(data=None, index: Optional[Collection[Any]] = None, columns: Optional[Collection[Any]] = None, dtype: Optional[Union[ExtensionDtype, str, dtype, Type[Union[str, float, int, complex, bool, object]]]] = None, copy: Optional[bool] = None)

Bases: DataFrame

Pydantic compatible data type for pandas DataFrame output.

classmethod validate(value: DataFrame) → Dict[str, Any]

Software Development Kit

Service API (daeploy.service)

Communication API (daeploy.communication)

Utilities API (daeploy.utilities)

Data Types API (daeploy.data_types)

Service API (`daeploy.service`)

Communication API (`daeploy.communication`)

Utilities API (`daeploy.utilities`)

Data Types API (`daeploy.data_types`)