malevich.square.utils#

class malevich.square.utils.Context#

Context contains all the necessary information about the run and the environment in which it is executed. Also, context provides a number of auxiliary functions for interacting with the environment, such as working with shared storage (share(), get_share_path(), delete_share()), dealing with common objects (common), access to the key-value storage (dag_key_value), and object storage (object_storage).

Usage#

Context object is used implicitly in apps. It may be requested by including an argument with explicit type annotation in the function signature:

from malevich.square import Context, processor

@processor()
def my_app(ctx: Context):
      pass

Example usage#

Here is some frequently used examples of context usage.

Accessing shared files#

To access shared files, you can use get_share_path() method.

from malevich.square import Context, processor, DF, APP_DIR

import pandas as pd
import os
from rembg import remove

def remove_background(
      images: DF['ImagePaths'],
      context: Context
):
      outputs = []
      for img in images.path_to_image.to_list():
         # A shared path (the one in data frame)
         # is only a key to a real file. The real
         # file is stored in a shared directory
         # and can be accessed using `get_share_path`
         img_file = context.get_share_path(
            img, not_exist_ok=True
         )

         img_array = cv2.imread(img_file)
         nobg = remove(
            img_array
         )

         # add _nobg before extention
         base, _ = os.path.splitext(img)
         base += "_nobg" + '.png'
         path = os.path.join(APP_DIR, _base)
         cv2.imwrite(
            path,
            nobg
         )

         # Sharing the file to pass it to the next app
         context.share(_base)
         outputs.append(_base)

      return pd.DataFrame(outputs, columns=['no_background_image'])

See the share() method for more details. Explore the code at GitHub.

app_cfg: Dict[str, Any]

App configuration. Given to the app at startup and contains arbitrary configuration data. The configuration is preserved between runs. This field is used to dictate the behaviour of the app.

Example:#

Assume, you have a processor that adds a column to a dataframe. You can configure the name of the column and its value using the app configuration:

from malevich.square import DF, Any, Context, processor


@processor()
def add_column(df: DF[Any], context: Context):
      # Using .get() method to have default values
      column_name = context.app_cfg.get('column', 'new_column')
      value = context.app_cfg.get('value', 'new_value')
      position = context.app_cfg.get('position', 0)

      # After configuration is parsed, we can add a column
      # to the dataframe according to it

      if position < 0:
         position = len(df.columns) + position + 1

      df.insert(position, column_name, value)

      return df

Source: GitHub.

Metascript#

When developing a flow in Metascript, you can pass the configuration using the config parameter:

from malevich import flow, collection
from malevich.utility import add_column

@flow()
def my_flow():
      data = collection('data.csv')
      add_column(data, config={
         'column': 'A',
         'value': '10',
         'position': -1
      })

app_id: str#: App ID (unique for each app). This ID given to an app at startup by interpreters.

run_id: str#: Run ID (unique for each run).

dag_key_value: _DagKeyValue#: Key-value storage. Shared between all apps of one run. Values must be bytes, string, int or float; dictionary order is not guaranteed

object_storage: _ObjectStorage#: Object storage, shared between all apps of one run.

common: Any#: An object shared between all runs of the app. It is not being explicitly serialized, so may be any object. In comparison, objects put into app_cfg can be used in the same way, but they are serialized and deserialized limiting the types that can be used. That is not the case for common.

share(path: str, all_runs: bool = False, path_prefix: str = '/julius/apps', force: bool = False, synchronize: bool = True) → None#

Shares a file or a directory between all apps for the current deployment.

The file or directory is copied to a shared directory, which is accessible by all apps. The file should be accessible by the following path:

{path_prefix}/{path}

where path_prefix and path are the corresponding arguments.

Note, that:

The path_prefix is not included into shared key.
If all_runs is set to True, the file or directory will be shared between all runs of the app. Otherwise, it will be shared only between the apps of the current run.
If synchronize is set to True, the file or directory will be copied to all apps. Otherwise, it will be copied only to apps with a common mount.
If the file or directory already exists in the shared directory, it will not be copied unless force is set to True.

Note

See Sharing files for more details.

Parameters:

path (str) – Key of shared file or directory
all_runs (bool, optional) – Whether to share the file or directory between all runs. Defaults to False.
path_prefix (str, optional) – Path prefix. Defaults to APP_DIR.
force (bool, optional) – Whether to overwrite the file or directory if it already exists. Defaults to False.
synchronize (bool, optional) – Whether to synchronize the file or directory between all apps. Defaults to True.

async async_share(path: str, all_runs: bool = False, path_prefix: str = '/julius/apps', force: bool = False, synchronize: bool = True) → None#

Shares a file or a directory between all apps for the current deployment.

It is an asynchronous version of share().

Parameters:

path (str) – Key of shared file or directory
all_runs (bool, optional) – Whether to share the file or directory between all runs. Defaults to False.
path_prefix (str, optional) – Path prefix. Defaults to APP_DIR.
force (bool, optional) – Whether to overwrite the file or directory if it already exists. Defaults to False.
synchronize (bool, optional) – Whether to synchronize the file or directory between all apps. Defaults to True.

share_many(paths: List[str], all_runs: bool = False, path_prefix: str = '/julius/apps', force: bool = False, synchronize: bool = True) → None#

Shares multiple files or directories.

The same as share(), but for multiple files or directories. Ignores paths that do not exist.

async async_share_many(paths: List[str], all_runs: bool = False, path_prefix: str = '/julius/apps', force: bool = False, synchronize: bool = True) → None#: Shares multiple files or directories. The same as async_share(), but for multiple files or directories. Ignores paths that do not exist.

get_share_path(path: str, all_runs: bool = False, not_exist_ok: bool = False) → str | None#

Retrieves a real file path by shared key.

Once file is shared, it is copied to a shared directory, which is accessible by all apps. However, to actually access the file, you have to retrieve its real path using in the mounted file system.

Note

See Accessing shared files for more details.

Parameters:

path (str) – Shared file key
all_runs (bool, optional) – To access files shared between all runs, set to True. Defaults to False.
not_exist_ok (bool, optional) – Whether to raise an exception if the file does not exist.

Returns:

Readable file path or None (if not_exist_ok is set to True)

Return type:

str

delete_share(path: str, all_runs: bool = False, synchronize: bool = True) → None#

Deletes previously shared file or directory.

Parameters:

path (str) – Shared file key
all_runs (bool, optional) – Whether to delete files shared between all runs. Defaults to False.
synchronize (bool, optional) – Whether to synchronize the deletion between all apps. Defaults to True.

async async_delete_share(path: str, all_runs: bool = False, synchronize: bool = True) → None#

Deletes previously shared file or directory.

Parameters:

path (str) – Shared file key
all_runs (bool, optional) – Whether to delete files shared between all runs. Defaults to False.
synchronize (bool, optional) – Whether to synchronize the deletion between all apps. Defaults to True.

synchronize(paths: List[str] | None = None, all_runs: bool = False) → None#

Forcefully synchronizes paths accross all apps.

Setting paths to None or an empty list will synchronize all shared files and directories.

Parameters:

paths (Optional[List[str]], optional) – Paths to synchronize.
all_runs (bool, optional) – Whether to synchronize files shared between all runs. Should be set to the same value as share() method was called with. Defaults to False.

async async_synchronize(paths: List[str] | None = None, all_runs: bool = False) → None#

Forcefully synchronizes paths accross all apps.

Setting paths to None or an empty list will synchronize all shared files and directories.

Parameters:

paths (Optional[List[str]], optional) – Paths to synchronize.
all_runs (bool, optional) – Whether to synchronize files shared between all runs. Should be set to the same value as share() method was called with. Defaults to False.

msg(data: str | Dict, url: str | None = None, headers: Dict[str, str] | None = None, wait: bool = False, wrap: bool = True, with_result: bool = False) → None#

Sends http msg to system or any url

Parameters:

data (Union[str, Dict]) – msg data
url (Optional[str]) – url to which the request is sent. If not specified, the default url is used (system url, but it can be overridden in the startup cfg, the parameter is msg_url). system url. Defaults to None.
headers (Optional[Dict[str, str]]) – Just headers, with which the request is sent. Defaults to None.
wait (bool) – wait result. Defaults to False.
wrap (bool) – need system wrap with operationId. Defaults to True.
with_result (bool) – return result. Defaults to False.

async async_msg(data: str | Dict, url: str | None = None, headers: Dict[str, str] | None = None, wait: bool = False, wrap: bool = True, with_result: bool = False)#

Sens http msg to system or any url

Parameters:

data (Union[str, Dict]) – msg data
url (Optional[str]) – url to which the request is sent. If not specified, the default url is used (system url, but it can be overridden in the startup cfg, the parameter is msg_url). system url. Defaults to None.
headers (Optional[Dict[str, str]]) – Just headers, with which the request is sent. Defaults to None.
wait (bool) – wait result. Defaults to False.
wrap (bool) – need system wrap with operationId. Defaults to True.
with_result (bool) – return result. Defaults to False.

email_send(message: str, subject: str | None = None, type: str = 'gmail') → None#

Send an email

Parameters:

message (str) – text message
subject (Optional[str], optional) – message subject, if subject is None used default. Defaults to None.
type (str, optional) – message type, only gmail work now. Defaults to “gmail”.

metadata(df_name: str) → Dict[str, Any] | List[Dict[str, Any]] | None#

Gets metadata by df_name (if it saved with collection)

Parameters:: df_name (str) – df name
Returns:: metadata if exists (list if many), None otherwise
Return type:: Optional[Union[Dict[str, Any], List[Dict[str, Any]]]]

property scale_info: Tuple[int, int]#

index and index count. index count - how many apps run it, index in [0, index count)

Returns:: index and index count
Return type:: Tuple[int, int]
Type:: Gets scale info

get_scale_part(df: DataFrame) → DataFrame#

Gets scale part of df (index and index count used for that) - all apps app get different data

Parameters:: df (pd.DataFrame) – df to scale
Returns:: scale part of df
Return type:: pd.DataFrame

property operation_id: str#

Operation identifier

Returns:: operation_id
Return type:: str

has_object(path: str) → bool#

Checks whether there is an asset (OBJ) at the path

Parameters:: path (str) – Path to object (asset)
Returns:: exist or not
Return type:: bool

get_object(path: str) → OBJ#

Composes a path to an asset (OBJ) and returns it as OBJ

If the asset (object) does not exist, the function fails with an error.

Parameters:: path (str) – Path to object (asset)
Returns:: An OBJ wrap around a path
Return type:: OBJ

as_object(paths: Dict[str, str], *, dir: str | None = None) → OBJ#

Creates an asset (OBJ) by copying paths to specific directory and creating OBJ by this dir.

Assets (OBJ) are simply a path within a directory accessible from within container by a certain user. This function creates a new asset by copying specified files into separate directory and creating an asset pointing to whole folder.

Parameters:

paths (Dict[str, str]) – Path to an actual object -> subpath in asset directory
dir (Optional[str], optional) – target directory name. If not set, it is generated. Defaults to None

Returns:

OBJ with created directory captured

Return type:

OBJ

class Context._ObjectStorage#

A storage for binary objects common to the user.

Works with cloud S3 storage and app shared file system. Provides an access to the objects shared with Context.share() method. and accessible with Context.get_share_path() method.

All methods of the object storage have local parameter. If it is set to True, the operation will be performed on the local file system. Otherwise, the operation will be performed on the remote object storage.

The information might be unsynchronized between apps. To ensure that the information is synchronized, use all_apps parameter. If it is set to True, the operation will be performed on all apps. Otherwise, the operation will be performed only on apps the method is called from.

get_keys(local: bool = False, all_apps: bool = False) → List[str]#

Get keys from local mount or remote object storage.

Parameters:

local (bool, optional) – whether to use local mount or remote object storage. If set to True, the operation will be performed on the local file system. Otherwise, the operation will be performed on the remote object storage. Defaults to False.
all_apps (bool, optional) – whether to synchronize the operation between all apps. If set to True, the operation will be performed on all apps. Otherwise, the operation will be performed only on apps the method is called from. Defaults to False.

Returns:

Keys from local mount or remote object storage.

Return type:

List[str]

async async_get_keys(local: bool = False, all_apps: bool = False) → List[str]#

Get keys from local mount or remote object storage.

Parameters:

local (bool, optional) – whether to use local mount or remote object storage. If set to True, the operation will be performed on the local file system. Otherwise, the operation will be performed on the remote object storage. Defaults to False.
all_apps (bool, optional) – whether to synchronize the operation between all apps. If set to True, the operation will be performed on all apps. Otherwise, the operation will be performed only on apps the method is called from. Defaults to False.

Returns:

Keys from local mount or remote object storage.

Return type:

List[str]

get(keys: List[str], force: bool = False, all_apps: bool = True) → List[str]#

Updates mount for this app (or all apps), return keys for which it was successful

Parameters:

keys (List[str]) – Keys by which values are obtained. If this is not possible, this key will not be returned in result)
force (bool, optional) – If set, it will ignore what is locally and load data from the remote object storage. Otherwise it will only take what does not exist. Defaults to False.
all_apps (bool, optional) – If set to true, the operation will be performed in all apps. Otherwise only for apps with associated mount. Defaults to True.

Returns:

Keys by which it was possible to obtain the value and load it into the mount

Return type:

List[str]

async async_get(keys: List[str], force: bool = False, all_apps: bool = True) → List[str]#

Updates mount for this app (or all apps), return keys for which it was successful

Parameters:

keys (List[str]) – Keys by which values are obtained. If this is not possible, this key will not be returned in result)
force (bool, optional) – If set, it will ignore what is locally and load data from the remote object storage. Otherwise it will only take what does not exist. Defaults to False.
all_apps (bool, optional) – If set to true, the operation will be performed in all apps. Otherwise only for apps with associated mount. Defaults to True.

Returns:

Keys by which it was possible to obtain the value and load it into the mount

Return type:

List[str]

get_all(local: bool = False, force: bool = False, all_apps: bool = True) → List[str]#

Updates mount and return all keys in it, if local - return result only for mount (or all apps mounts if all_apps), otherwise - load all by remote object storage

Parameters:

local (bool, optional) – whether to use local mount or remote object storage. If set to True, the operation will be performed on the local file system. Otherwise, the operation will be performed on the remote object storage. Defaults to False.
force (bool, optional) – If set, it will ignore what is locally and load data from the remote object storage. Otherwise it will only take what does not exist. Defaults to False.
all_apps (bool, optional) – If set to true, the operation will be performed in all apps. Otherwise only for apps with associated mount. Defaults to True.

Returns:

All keys in the mount or all apps mounts if all_apps is True, otherwise load all keys from remote object storage.

Return type:

List[str]

async async_get_all(local: bool = False, force: bool = False, all_apps: bool = True) → List[str]#

Updates mount and return all keys in it, if local - return result only for mount (or all apps mounts if all_apps), otherwise - load all by remote object storage

Parameters:

local (bool, optional) – whether to use local mount or remote object storage. If set to True, the operation will be performed on the local file system. Otherwise, the operation will be performed on the remote object storage. Defaults to False.
force (bool, optional) – If set, it will ignore what is locally and load data from the remote object storage. Otherwise it will only take what does not exist. Defaults to False.
all_apps (bool, optional) – If set to true, the operation will be performed in all apps. Otherwise only for apps with associated mount. Defaults to True.

Returns:

All keys in the mount or all apps mounts if all_apps is True, otherwise load all keys from remote object storage.

Return type:

List[str]

update(keys: List[str], presigned_expire: int | None = -1) → Dict[str, str]#

Updates objects in remote storage

Retrieves objects from local mount and updates remote object storage. If presigned_expire is set to a positive value, creates and returns presigned urls for the objects.

If presigned_expire is None, set to default timeout.

If presigned_expire is negative, returns an empty dictionary.

Parameters:

keys (List[str]) – Keys to update
presigned_expire (int, optional) – If positve, life span of presigned urls in seconds. If None, set to default timeout. Defaults to -1.

Returns:

Mapping of keys to presigned urls

Return type:

Dict[str, str]

async async_update(keys: List[str], presigned_expire: int | None = -1) → Dict[str, str]#

Updates objects in remote storage

Retrieves objects from local mount and updates remote object storage. If presigned_expire is set to a positive value, creates and returns presigned urls for the objects.

If presigned_expire is None, set to default timeout.

If presigned_expire is negative, returns an empty dictionary.

Parameters:

keys (List[str]) – Keys to update
presigned_expire (int, optional) – If positve, life span of presigned urls in seconds. If None, set to default timeout. Defaults to -1.

Returns:

key to presigned url

Return type:

Dict[str, str]

presigned(keys: List[str], expire: int | None = None) → Dict[str, str]#

Creates presigned urls for specified keys

Parameters:

keys (List[str]) – Keys to create presigned urls for
expire (int, optional) – Life span of presigned urls in seconds (must be positive). If None, set to default timeout. Defaults to None.

Returns:

Mapping of keys to presigned urls

Return type:

Dict[str, str]

async async_presigned(keys: List[str], expire: int | None = None) → Dict[str, str]#

Creates presigned urls for specified keys

Parameters:

keys (List[str]) – Keys to create presigned urls for
expire (int, optional) – Life span of presigned urls in seconds (must be positive). If None, set to default timeout. Defaults to None.

Returns:

Mapping of keys to presigned urls

Return type:

Dict[str, str]

delete(keys: List[str]) → None#

Deletes values in mount and remote storage

Parameters:: keys (List[str]) – Keys to delete

async async_delete(keys: List[str]) → None#

Deletes values in mount and remote storage

Parameters:: keys (List[str]) – Keys to delete

class Context._DagKeyValue(run_id: str | None = None)#

Simple key-value storage, shared for all apps of one run. Values must be bytes, string, int or float; dictionary order is not guaranteed

get_bytes(key: str) → bytes#

Gets a binary value by key more optimally.

Consider using this to retrieve binary data in more efficient way, but keep in mind that get() can also be used

Parameters:: key (str) – key in storage
Returns:: Value stored by key
Return type:: bytes

async async_get_bytes(key: str) → bytes#

Gets a binary value by key more optimally.

Consider using this to retrieve binary data in more efficient way, but keep in mind that get() can also be used

Parameters:: key (str) – key in storage
Returns:: Value stored by key
Return type:: bytes

get(keys: List[str]) → Dict[str, Any]#

Gets values by keys

Retrieves a slice of storage by keys. If a key is not found, the value will be set to None

Parameters:: keys (list[str]) – list of keys
Returns:: A dictionary of key-value pairs.
Return type:: Dict[str, Any]

async async_get(keys: List[str]) → Dict[str, Any]#

Gets values by keys

Retrieves a slice of storage by keys. If a key is not found, the value will be set to None

Parameters:: keys (list[str]) – list of keys
Returns:: A dictionary of key-value pairs.
Return type:: Dict[str, Any]

get_all() → Dict[str, Any]#

Gets all values

Retrieves the whole storage as a dictionary.

Returns:: A dictionary of key-value pairs.
Return type:: Dict[str, Any]

async async_get_all() → Dict[str, Any]#

Gets all values

Retrieves the whole storage as a dictionary.

Returns:: A dictionary of key-value pairs.
Return type:: Dict[str, Any]

update(keys_values: Dict[str, Any]) → None#

Sets values by keys

Accepts a dictionary of key-value pairs and sets them in storage. If a key already exists, it will be overwritten.

Parameters:: keys_values (dict) – A dictionary of key-value pairs.

async async_update(keys_values: Dict[str, Any]) → None#

Sets values by keys

Accepts a dictionary of key-value pairs and sets them in storage. If a key already exists, it will be overwritten.

Parameters:: keys_values (dict) – A dictionary of key-value pairs.

clear() → None#: Purges the storage

async async_clear() → None#: Purges the storage

class malevich.square.utils.S3Helper(client: Any, s3_bucket: str)#

Ready-made auxiliary wrapper for interacting with custom s3

static create_by_cfg(cfg: Dict[str, Any], **kwargs) → S3Helper#

get_object(key: str, bucket: str | None = None) → StreamingBody | None#

Uses get_object from client by bucket and key

Parameters:

key (str) – object storage key
bucket (Optional[str], optional) – object storage bucket, if set - override default. Defaults to None.

Returns:

get_object response body

Return type:

Optional[StreamingBody]

get_df(key: str, bucket: str | None = None) → DataFrame#

Uses get_object result and cast it to data frame

Parameters:

key (str) – object storage key
bucket (Optional[str], optional) – object storage bucket, if set - override default. Defaults to None.

Returns:

result in df view

Return type:

pd.DataFrame

save_object(body: Any, key: str, bucket: str | None = None) → None#

Uses put_object from client by bucket, key and body

Parameters:

body (Any) – saved data
key (str) – object storage key
bucket (Optional[str], optional) – object storage bucket, if set - override default. Defaults to None.

save_df(df: DataFrame, key: str, bucket: str | None = None) → None#

Uses save_object to save df by bucket and key

Parameters:

df (pd.DataFrame) – df to save
key (str) – object storage key
bucket (Optional[str], optional) – object storage bucket, if set - override default. Defaults to None.

delete_object(key: str, bucket: str | None = None) → None#

Deletes object by bucket and key

Parameters:

key (str) – object storage key
bucket (Optional[str], optional) – object storage bucket, if set - override default. Defaults to None.

class malevich.square.utils.SmtpSender(login: str, password: str, smtp_server: str = 'smtp.gmail.com', smtp_port: int = 465)#

Ready-made auxiliary wrapper for interacting with SMTP :param login: login :type login: str :param password: password :type password: str :param smtp_server: smtp server. Defaults to “smtp.gmail.com”. :type smtp_server: str, optional :param smtp_port: smtp port. Defaults to 465. :type smtp_port: int, optional

send(receivers: list[str], subject: str, message: str) → None#

Sends an email

Parameters:

receivers (list[str]) – list of emails
subject (str) – message subject
message (str) – message text

malevich.square.utils.to_df(x: Any, force: bool = False) → DataFrame#

Creates a data frame from an arbitrary object - torch.Tensor: Tensor is serialized using torch.save and then encoded using base112. Autograd information is preserved. - numpy, list, tuple, range, bytearray: Data is serialized using pickle and stored as is in data column. - set, frozenset: Data is converted to list and stored as is in data column. - dict: Data is serialized using json and stored as is in data column. - int, float, complex, str, bytes, bool: Data is stored as is in data column.

Parameters:

x (Any) – Object to convert to data frame
force (bool, optional) – If set, it will ignore the type of the object and serialize it using pickle. Defaults to False.

Returns:

Data frame with a single column data

Return type:

pd.DataFrame

malevich.square.utils.from_df(x: DataFrame, type_name: str | None = None, force: bool = False) → Any#

Converts a data frame obtained by running to_df() back to an object

Parameters:

x (pd.DataFrame) – Data frame to convert
type_name (Optional[str], optional) – Type of the object to convert to. If not specified, the type is inferred from the data frame. Defaults to None.
force (bool, optional) – If set, it will ignore the type of the object and deserialize it using pickle. Defaults to False.

Returns:

Object of type type_name or inferred type

Return type:

Any

malevich.square.utils.to_binary(smth: Any) → bytes#: Converts object to binary :param smth: object to convert :type smth: Any

malevich.square.utils.from_binary(smth: bytes) → Any#: Converts binary to object :param smth: binary to convert :type smth: bytes

malevich.square.utils.APP_DIR = '/julius/apps'#: Working directory from which the app is run. Equivalent to os.getcwd() from within the app.

malevich.square.utils.WORKDIR = '/julius'#: Directory into which the user code is copied during app construction.

malevich.square.utils#

Usage#

Example usage#

Sharing files#

Accessing shared files#

Example:#

Metascript#