appmesh_client module

App Mesh Python SDK

class appmesh_client.App(data=None)[source]

Bases: object

App object present an application in App Mesh

Construct an App Mesh Application object

Parameters:: data (str | dict | json, optional) – application definition data

class Behavior(data=None)[source]

Bases: object

Application error handling behavior definition object

class Action(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)[source]

Bases: Enum

Keepalive = 'keepalive'

Remove = 'remove'

Restart = 'restart'

Standby = 'standby'

set_control_behavior(control_code: int, a: Action) → None[source]: Set error handling behavior while application exit with specific return code

set_exit_behavior(a: Action) → None[source]: Set error handling behavior while application exit

class DailyLimitation(data=None)[source]

Bases: object

Application avialable day time definition object

set_daily_range(start: datetime, end: datetime) → None[source]: Set valid day hour range

class Permission(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)[source]

Bases: Enum

Deny = '1'

Read = '2'

Write = '3'

class ResourceLimitation(data=None)[source]

Bases: object

Application cgroup limitation definition object

json()[source]

set_env(k: str, v: str, secure: bool = False) → None[source]: Set environment variable

set_permission(group_user: Permission, others_user: Permission) → None[source]

set_valid_time(start: datetime, end: datetime) → None[source]: Set avialable time window

class appmesh_client.AppMeshClient(rest_url: str = 'https://127.0.0.1:6060', rest_ssl_verify=False, rest_ssl_client_cert=None, rest_timeout=(60, 300), jwt_token=None)[source]

Bases: object

Client object used to access App Mesh REST Service

install pip package: python3 -m pip install –upgrade appmesh
import module: from appmesh import appmesh_client

Construct an App Mesh client object

Parameters:

rest_url (str, optional) – server URI string.
rest_ssl_verify (str, optional) – (optional) SSL CA certification. Either a boolean, in which case it controls whether we verify the server’s TLS certificate, or a string, in which case it must be a path to a CA bundle to use. Defaults to True.
rest_ssl_client_cert (tuple, optional) – SSL client certificate and key pair. If String, path to ssl client cert file (.pem). If Tuple, (‘cert’, ‘key’) pair.
rest_timeout (tuple, optional) – HTTP timeout, Defaults to 60 seconds for connect timeout and 300 seconds for read timeout
jwt_token (str, optional) – JWT token, provide correct token is same with login() & authenticate().

class Method(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)[source]

Bases: Enum

REST methods

DELETE = 'DELETE'

GET = 'GET'

POST = 'POST'

POST_STREAM = 'POST_STREAM'

PUT = 'PUT'

app_add(app: App) → App[source]

Register an application

Parameters:: app (App) – the application definition.
Returns:: resigtered application object.
Return type:: App

Exception:: failed request

app_delete(app_name: str)[source]

Remove an application.

Parameters:: app_name (str) – the application name.
Returns:: success or failure. str: text message.
Return type:: bool

app_disable(app_name: str)[source]

Stop and disable an application

Parameters:: app_name (str) – the application name.
Returns:: success or failure. str: text message.
Return type:: bool

app_enable(app_name: str)[source]

Enable an application

Parameters:: app_name (str) – the application name.
Returns:: success or failure. str: text message.
Return type:: bool

app_health(app_name: str)[source]

Get application health status, 0 is health.

Parameters:: app_name (str) – the application name.
Returns:: ‘0’ is heathy, ‘1’ is unhealthy.
Return type:: str

app_output(app_name: str, stdout_position: int = 0, stdout_index: int = 0, stdout_maxsize: int = 10240, process_uuid: str = '', timeout: int = 0)[source]

Get application stdout/stderr

Parameters:

app_name (str) – the application name
stdout_position (int, optional) – start read position, 0 means start from beginning.
stdout_index (int, optional) – index of history process stdout, 0 means get from current running process, the stdout number depends on ‘stdout_cache_size’ of the application.
stdout_maxsize (int, optional) – max buffer size to read.
process_uuid (str, optional) – used to get the specified process.
timeout (int, optional) – wait for the running process for some time(seconds) to get the output.

Returns:

success or failure. str: output string. int or None: current read position. int or None: process exit code.

Return type:

bool

app_view(app_name: str) → App[source]

Get one application information

Parameters:: app_name (str) – the application name.
Returns:: the application object both contain static configuration and runtime information.
Return type:: App

Exception:: failed request or no such application

app_view_all()[source]

Get all applications

Returns:: the application object both contain static configuration and runtime information, only return applications that the user has permissions.
Return type:: list

Exception:: failed request or no such application

authentication(token: str, permission=None) → bool[source]

Login with token and verify permission when specified

Parameters:

token (str) – JWT token returned from login().
permission (str, optional) – the permission ID used to verify the token user permission ID can be: - pre-defined by App Mesh from security.json (e.g ‘app-view’, ‘app-delete’) - defined by input from role_update() or security.json

Returns:

authentication success or failure.

Return type:

bool

cloud_app(app_name: str)[source]

Get an cloud application

Parameters:: app_name (str) – the application name.
Returns:: success or failure. dict: application in JSON format.
Return type:: bool

cloud_app_add(app_json: dict)[source]

Add a cloud application

Parameters:: app_json – the cloud application definition with replication, condition and resource requirement

cloud_app_delete(app_name: str) → bool[source]

Delete a cloud application

Parameters:: app_name (str) – The application name for cloud
Returns:: success or failure.
Return type:: bool

cloud_app_output(app_name: str, host_name: str, stdout_position: int = 0, stdout_index: int = 0, stdout_maxsize: int = 10240, process_uuid: str = '')[source]

Get cloud application stdout/stderr from master agent

Parameters:

app_name (str) – the application name
host_name (str) – the target host name where the application is running
stdout_position (int, optional) – start read position, 0 means start from beginning.
stdout_index (int, optional) – index of history process stdout, 0 means get from current running process, the stdout number depends on ‘stdout_cache_size’ of the application.
stdout_maxsize (int, optional) – max buffer size to read.
process_uuid (str, optional) – used to get the specified process.

Returns:

success or failure. str: output string. int or None: current read position. int or None: process exit code.

Return type:

bool

cloud_app_view_all()[source]

Get all cloud applications

Returns:: success or failure. dict: cloud applications in JSON format.
Return type:: bool

cloud_nodes()[source]

Get cluster node list

Returns:: success or failure. dict: cluster node list json.
Return type:: bool

config_set(cfg_json)[source]

Update configuration, the format follow ‘config.json’, support partial update

Parameters:: cfg_json (dict) – the new configuration json.
Returns:: success or failure. dict: the updated configuration json.
Return type:: bool

config_view()[source]

Get App Mesh configuration JSON

Returns:: success or failure. dict: the configuration json.
Return type:: bool

file_download(file_path: str, local_file: str) → bool[source]

Copy a remote file to local, the local file will have the same permission as the remote file

Parameters:

file_path (str) – the remote file path.
local_file (str) – the local file path to be downloaded.

Returns:

success or failure.

Return type:

bool

file_upload(local_file: str, file_path: str)[source]

Upload a local file to the remote server, the remote file will have the same permission as the local file

Dependency:: sudo apt install python3-pip pip3 install requests_toolbelt

Parameters:

local_file (str) – the local file path.
file_path (str) – the target remote file to be uploaded.

Returns:

success or failure. str: text message.

Return type:

bool

groups_view()[source]

Get all user groups

Returns:: success or failure. dict: user group array.
Return type:: bool

host_resource()[source]

Get App Mesh host resource report include CPU, memory and disk

Returns:: success or failure. dict: the host resource json.
Return type:: bool

property jwt_token: str

property for jwt_token

Returns:: _description_
Return type:: str

log_level_set(level: str = 'DEBUG')[source]

Update App Mesh log level(DEBUG/INFO/NOTICE/WARN/ERROR), a wrapper of config_set()

Parameters:: level (str, optional) – log level.
Returns:: success or failure. dict: the updated configuration json.
Return type:: bool

login(user_name: str, user_pwd: str, totp_code='', timeout_seconds='P1W') → str[source]

Login with user name and password

Parameters:

user_name (str) – the name of the user.
user_pwd (str) – the password of the user.
totp_code (str, optional) – the TOTP code if enabled for the user.
timeout_seconds (int | str, optional) – token expire timeout of seconds. support ISO 8601 durations (e.g., ‘P1Y2M3DT4H5M6S’ ‘P1W’).

Returns:

JWT token if verify success, otherwise return None.

Return type:

str

logoff() → bool[source]

Logoff current session from server

Returns:: logoff success or failure.
Return type:: bool

metrics()[source]

Prometheus metrics (this does not call Prometheus API /metrics, just copy the same metrics data)

Returns:: success or failure. str: prometheus metrics texts
Return type:: bool

permissions_for_user()[source]

Get current user permissions

Returns:: success or failure. dict: user permission array.
Return type:: bool

permissions_view()[source]

Get all available permissions

Returns:: success or failure. dict: permission array.
Return type:: bool

renew(timeout_seconds='P1W') → str[source]

Renew current token

Parameters:: timeout_seconds (int | str, optional) – token expire timeout of seconds. support ISO 8601 durations (e.g., ‘P1Y2M3DT4H5M6S’ ‘P1W’).
Returns:: The new JWT token if renew success, otherwise return None.
Return type:: str

role_delete(role_name: str) → bool[source]

Delete a user role

Parameters:: role_name (str) – the role name.
Returns:: success or failure.
Return type:: bool

role_update(role_name: str, role_permission_json: dict) → bool[source]

Update (or add) a role with defined permissions, the permission ID can be App Mesh pre-defined or other permission ID.

Parameters:

role_name (str) – the role name.
role_permission_json (dict) – role permission definition array, e.g: [“app-control”, “app-delete”, “cloud-app-reg”, “cloud-app-delete”]

Returns:

success or failure.

Return type:

bool

roles_view()[source]

Get all roles with permission definition

Returns:: success or failure. dict: all role definition.
Return type:: bool

run_async(app: App, max_time_seconds='PT1H', life_cycle_seconds='PT10H')[source]

Asyncrized run a command remotely, ‘name’ attribute in app_json dict used to run an existing application Asyncrized run will not block process

Parameters:

app (App) – application object.
max_time_seconds (int | str, optional) – max run time for the remote process, support ISO 8601 durations (e.g., ‘P1Y2M3DT4H5M6S’ ‘P5W’).
life_cycle_seconds (int | str, optional) – max lifecycle time for the remote process. support ISO 8601 durations (e.g., ‘P1Y2M3DT4H5M6S’ ‘P5W’).

Returns:

app_name, new application name for this run str: process_uuid, process UUID for this run

Return type:

str

run_async_wait(run: Run, stdout_print: bool = True, timeout: int = 0) → int[source]

Wait for an async run to be finished

Parameters:

run (Run) – asyncrized run result from run_async().
stdout_print (bool, optional) – print remote stdout to local or not.
timeout (int, optional) – wait max timeout seconds and return if not finished, 0 means wait until finished

Returns:

return exit code if process finished, return None for timeout or exception.

Return type:

int

run_sync(app: App, stdout_print: bool = True, max_time_seconds='PT1H', life_cycle_seconds='PT10H') → int[source]

Block run a command remotely, ‘name’ attribute in app_json dict used to run an existing application The synchronized run will block the process until the remote run is finished then return the result from HTTP response

Parameters:

app (App) – application object.
stdout_print (bool, optional) – whether print remote stdout to local or not. Defaults to True.
max_time_seconds (int | str, optional) – max run time for the remote process. support ISO 8601 durations (e.g., ‘P1Y2M3DT4H5M6S’ ‘P5W’).
life_cycle_seconds (int | str, optional) – max lifecycle time for the remote process. support ISO 8601 durations (e.g., ‘P1Y2M3DT4H5M6S’ ‘P5W’).

Returns:

process exit code, return None if no exit code.

Return type:

int

tag_add(tag_name: str, tag_value: str) → bool[source]

Add a new label

Parameters:

tag_name (str) – the label name.
tag_value (str) – the label value.

Returns:

success or failure.

Return type:

bool

tag_delete(tag_name: str) → bool[source]

Delete a label

Parameters:: tag_name (str) – the label name.
Returns:: success or failure.
Return type:: bool

tag_view()[source]

Get the server labels

Returns:: success or failure. dict: label data.
Return type:: bool

totp_disable(user='self') → bool[source]

Disable 2FA for current user

Parameters:: user (str, optional) – user name for disable TOTP.
Returns:: success or failure.
Return type:: bool

totp_secret() → str[source]

Generate TOTP secret for current login user and return MFA URI with JSON body

Returns:: TOTP secret str
Return type:: str

totp_setup(totp_code: str) → bool[source]

Setup 2FA for current login user

Parameters:: totp_code (str) – TOTP code
Returns:: success or failure.
Return type:: bool

user_add(user_name: str, user_json: dict) → bool[source]

Add a new user, not available for LDAP user

Parameters:

user_name (str) – the user name.
user_json (dict) – user definition, follow same user format from security.json.

Returns:

success or failure.

Return type:

bool

user_delete(user_name: str) → bool[source]

Delete a user

Parameters:: user_name (str) – the user name.
Returns:: success or failure.
Return type:: bool

user_lock(user_name: str) → bool[source]

Lock a user

Parameters:: user_name (str) – the user name.
Returns:: success or failure.
Return type:: bool

user_passwd_update(new_password: str, user_name: str = 'self')[source]

Change user password

Parameters:

user_name (str) – the user name.
new_password (str) – the new password string

Returns:

success or failure. str: result message.

Return type:

bool

user_self()[source]

Get current user infomation

Returns:: success or failure. dict: user definition.
Return type:: bool

user_unlock(user_name: str) → bool[source]

Unlock a user

Parameters:: user_name (str) – the user name.
Returns:: success or failure.
Return type:: bool

users_view()[source]

Get all users

Returns:: success or failure. dict: all user definition.
Return type:: bool

class appmesh_client.AppMeshClientTCP(rest_ssl_verify=False, rest_ssl_client_cert=None, jwt_token=None, tcp_address=('localhost', 6059))[source]

Bases: AppMeshClient

Client object used to access App Mesh REST Service over TCP (better performance than AppMeshClient)

Dependency:: pip3 install msgpack

Construct an App Mesh client TCP object

Parameters:

rest_ssl_verify (str, optional) – (optional) SSL CA certification. Either a boolean, in which case it controls whether we verify the server’s TLS certificate, or a string, in which case it must be a path to a CA bundle to use. Defaults to True.
rest_ssl_client_cert (tuple, optional) – SSL client certificate and key pair. If String, path to ssl client cert file (.pem). If Tuple, (‘cert’, ‘key’) pair.
jwt_token (str, optional) – JWT token, provide correct token is same with login() & authenticate().
tcp_address (tuple, optional) – TCP connect address.

file_download(file_path: str, local_file: str) → bool[source]

Copy a remote file to local, the local file will have the same permission as the remote file

Parameters:

file_path (str) – the remote file path.
local_file (str) – the local file path to be downloaded.

Returns:

success or failure.

Return type:

bool

file_upload(local_file: str, file_path: str)[source]

Upload a local file to the remote server, the remote file will have the same permission as the local file

Dependency:: sudo apt install python3-pip pip3 install requests_toolbelt

Parameters:

local_file (str) – the local file path.
file_path (str) – the target remote file to be uploaded.

Returns:

success or failure. str: text message.

Return type:

bool

class appmesh_client.Run(client, app_name: str, process_id: str)[source]

Bases: object

Application run object indicate to a remote run from run_async()

wait(stdout_print: bool = True, timeout: int = 0) → int[source]

Wait for an async run to be finished

Parameters:

run (Run) – asyncrized run result from run_async().
stdout_print (bool, optional) – print remote stdout to local or not.
timeout (int, optional) – wait max timeout seconds and return if not finished, 0 means wait until finished

Returns:

return exit code if process finished, return None for timeout or exception.

Return type:

int