Pipeline File Import

This module introduces tools that allow importing Pipeline objects from json/yaml files.

• JSONImporter is a class that imports pipeline from files

• get_chatsky_objects() is a function that provides an index of objects commonly used in a Pipeline definition.

exception JSONImportError

Bases: Exception

An exception for incorrect usage of JSONImporter.

class JSONImporter(custom_dir)

Bases: object

Enables pipeline import from file.

Since Pipeline and all its components are already pydantic BaseModel, the only purpose of this class is to allow importing and instantiating arbitrary objects.

Import is done by replacing strings of certain patterns with corresponding objects. This process is implemented in resolve_string_reference().

Instantiating is done by replacing dictionaries where a single key is an imported object with an initialized object where arguments are specified by the dictionary values. This process is implemented in replace_resolvable_objects() and parse_args().

Parameters:

custom_dir (Union[str, Path]) – Path to the directory containing custom code available for import under the CUSTOM_DIR_NAMESPACE_PREFIX.

CHATSKY_NAMESPACE_PREFIX: str = 'chatsky.'

Prefix that indicates an import from the chatsky library.

This class variable can be changed to allow using a different prefix.

CUSTOM_DIR_NAMESPACE_PREFIX: str = 'custom.'

Prefix that indicates an import from the custom directory.

This class variable can be changed to allow using a different prefix.

EXTERNAL_LIB_NAMESPACE_PREFIX: str = 'external:'

Prefix that indicates an import from any library.

This class variable can be changed to allow using a different prefix.

static is_resolvable(value)

Check if value starts with any of the namespace prefixes:

• CHATSKY_NAMESPACE_PREFIX;

• CUSTOM_DIR_NAMESPACE_PREFIX;

• EXTERNAL_LIB_NAMESPACE_PREFIX.

Return type:

bool

Returns:

Whether the value should be resolved (starts with a namespace prefix).

static sys_path_append(path)

Append path to sys.path before yielding and restore sys.path to initial state after returning.

static replace_prefix(string, old_prefix, new_prefix)

Replace old_prefix in string with new_prefix.

Raises:

ValueError – If the string does not begin with old_prefix.

Return type:

str

Returns:

A new string with a new prefix.

resolve_string_reference(obj)

Import an object indicated by obj.

First, obj is pre-processed – prefixes are replaced to allow import:

• CUSTOM_DIR_NAMESPACE_PREFIX is replaced {stem}. where stem is the stem of the custom dir;

• CHATSKY_NAMESPACE_PREFIX is replaced with chatsky.;

• EXTERNAL_LIB_NAMESPACE_PREFIX is removed.

Next the resulting string is imported: If the string is a.b.c.d, the following is tried in order:

1. from a import b; return b.c.d

2. from a.b import c; return c.d

3. from a.b.c import d; return d

For custom dir imports; parent of the custom dir is appended to sys.path via sys_path_append().

Return type:

Any

Returns:

An imported object.

Raises:

• ValueError – If obj does not begin with any of the prefixes (is not is_resolvable()).

• JSONImportError – If a string could not be imported. Includes exceptions raised on every import attempt.

parse_args(value)

Parse value into args and kwargs:

• If value is a dictionary, it is returned as kwargs;

• If value is a list, it is returned as args;

• If value is None, both args and kwargs are empty;

• If value is anything else, it is returned as the only arg.

Return type:

Tuple[list, dict]

Returns:

A tuple of args and kwargs.

replace_resolvable_objects(obj)

Replace any resolvable objects inside obj with their resolved versions and initialize any that are the only key of a dictionary.

This method iterates over every value inside obj (which is JsonValue). Any string that is_resolvable() is replaced with an object return from resolve_string_reference(). This is done only once (i.e. if a string is resolved to another resolvable string, that string is not resolved).

Any dictionaries that contain only one resolvable key are replaced with a result of resolve_string_reference(key)(*args, **kwargs) (the object is initialized) where args and kwargs is the result of parse_args() on the value of the dictionary.

Return type:

Any

Returns:

A new object with replaced resolvable strings and dictionaries.

import_pipeline_file(file)

Import a dictionary from a json/yaml file and replace resolvable objects in it.

Return type:

dict

Returns:

A result of replace_resolvable_objects() on the dictionary.

Raises:

• JSONImportError – If a file does not have a correct file extension.

• JSONImportError – If an imported object from file is not a dictionary.

get_chatsky_objects()

Return an index of most commonly used chatsky objects (in the context of pipeline initialization).

Returns:

A dictionary where keys are names of the objects (e.g. chatsky.core.Message) and values are the objects. The items in the dictionary are all the objects from the __init__ files of the following modules:

• ”chatsky.cnd”;

• ”chatsky.rsp”;

• ”chatsky.dst”;

• ”chatsky.proc”;

• ”chatsky.core”;

• ”chatsky.core.service”;

• ”chatsky.slots”;

• ”chatsky.context_storages”;

• ”chatsky.messengers”.