ai_core_sdk.content.workflow
index
/home/jenkins/agent/workspace/AI-Foundation_ai-core-sdk_master/ai_core_sdk/content/workflow.py

 
Modules
       
ast
pathlib
asyncio.subprocess
yaml

 
Classes
       
builtins.object
Workflow
enum.Enum(builtins.object)
WorkflowType

 
class Workflow(builtins.object)
    Workflow(name: str, path: pathlib.Path, type: ai_core_sdk.content.workflow.WorkflowType, content_package_version: Optional[str] = None, class_name: Optional[str] = None, dockerfiles: Dict[str, pathlib.Path] = None, description: str = <factory>, additional_options: Dict[str, Any] = <factory>) -> None
 
The Workflow object representing a workflow.
Usually instances of this class are not created manually but are created from a
class:`ai_core_sdk.content.content_package.ContentPackage` instance.
:param name: Name of the workflow
:type name: str
:param path: Path to the workflow file. Type of the file depends on the worklfow type.
:type path: pathlib.Path
:param type: Type of the workflow
:type type: class:`ai_core_sdk.content.workflow.WorkflowType`
:param content_package_version: Version of the content package
:type content_package_version: Union[str, None], optional
:param class_name: Name of the content package's class_name
:type class_name: Union[str, None], optional
:param description: Description text
:type description: str, optional
:param additional_options: Version str
:type additional_options: Dict[str, Any], optional
 
  Methods defined here:
__eq__(self, other)
Return self==value.
__init__(self, name: str, path: pathlib.Path, type: ai_core_sdk.content.workflow.WorkflowType, content_package_version: Optional[str] = None, class_name: Optional[str] = None, dockerfiles: Dict[str, pathlib.Path] = None, description: str = <factory>, additional_options: Dict[str, Any] = <factory>) -> None
Initialize self.  See help(type(self)) for accurate signature.
__repr__(self)
Return repr(self).
__str__(self)
Return str(self).
create_image(self, workflow_config: Union[str, pathlib.Path, Dict], docker_type: Optional[str] = None, return_cmd: bool = False, silent: bool = False)
Create docker image based on workflow_config.
The image is created using the tag workflow_config['image'].
:param workflow_config: Path to the workflow config or dictionary from a loaded workflow config
:type workflow_config: Union[str, pathlib.Path, Dict]
:param dockertype: Specific docker type, if not set the first in the dockerfiles definition is being used.
:type dockertype: str, optional
:param return_cmd: Return the command instead of executing it
:type return_cmd: str, optional
:raises: class:`CalledProcessError`, class:`AttributeError`
:return: Command as list of str or None
:rtype: Union[List[str], None]
create_template(self, workflow_config: Union[str, pathlib.Path, Dict, NoneType] = None, target_file: Union[str, pathlib.Path, NoneType] = None, return_cmd: bool = False, silent: bool = False) -> Optional[List[str]]
Create docker image based on workflow_config.
The image is created using the tag workflow_config['image'].
:param workflow_config: Path to the workflow config or dictionary from a loaded workflow
config. Can also be None for DeploymentYaml workflows.
:type workflow_config: Union[str, pathlib.Path, Dict, NOne]
:param target_file: Output path for the template. If None the template will be printed.
:type target_file: Union[str, pathlib.Path, None], optional
:param return_cmd: Return the command instead of executing it
:type return_cmd: str, optional
:raises: class:`CalledProcessError`, class:`AttributeError`
:return: Command as list of str or None
:rtype: Union[List[str], None]
raw_build_docker(self, *args: str, docker_type: Optional[str] = None, return_cmd: bool = False, silent: bool = False) -> Optional[List[str]]
Execute docker build CLI command.
This functions is to execute docker build CLI commands on dockerfile
of workflow. The basic structure of the CLI commands is:
`docker build -f <workflow.dockerfile[dockertype]> --build-arg pkg_version===<content_pkg_version> <base folder of the workflow>`.
:param `*args`: CLI arguments as strings
:type `*args`: str
:param dockertype: Specific docker type, if not set the first in the dockerfiles definition is being used.
:type dockertype: str, optional
:param return_cmd: Return the command instead of executing it
:type return_cmd: str, optional
:raises: class:`CalledProcessError`
:return: Command as list of str or None
:rtype: Union[List[str], None]
raw_metaflow_cli(self, *args: str, return_cmd: bool = False) -> Optional[str]
Execute metaflow CLI command.
This functions is to execute metaflow CLI commands on a workflow
python file. The basic structure of the CLI commands is:
`cd <base folder of the workflow> && python -m<flow.py> *args`.
:param `*args`: CLI arguments as strings
:type `*args`: str
:param return_cmd: Return the command instead of executing it
:type return_cmd: bool, optional
:raises: class:`CalledProcessError`
:return: Command as list of str or None
:rtype: Union[List[str], None]
validate(self) -> None
Run validationfor the content package definition and its workflows.
:raises: class:`ValueError`

Class methods defined here:
from_entry(name: str, entry: Dict[str, Any], content_package_version: Optional[str] = None) -> 'Workflow' from builtins.type
Create a worfklow from dictionary.
Usually instances of this class are not created manually but are created from a
class:`ai_core_sdk.content.content_package.ContentPackage` instance.
:param name: Name of the workflow
:type name: str
:param entry: Path to the workflow file. Type of the file depends on the worklfow type.
:type entry: Dict[str, Any]
:param content_package_version: Version of the content package
:type content_package_version: Union[str, None], optional
:raises: class:`KeyError`
:return: Workflow
:rtype: class:`ai_core_sdk.content.workflow.Workflow`

Readonly properties defined here:
absolute_path
Return absolute path of the workflow file (py/yaml).
:return: Path
:rtype: class:`pathlib.Path`

Data descriptors defined here:
__dict__
dictionary for instance variables (if defined)
__weakref__
list of weak references to the object (if defined)

Data and other attributes defined here:
__annotations__ = {'additional_options': typing.Dict[str, typing.Any], 'class_name': typing.Optional[str], 'content_package_version': typing.Optional[str], 'description': <class 'str'>, 'dockerfiles': typing.Dict[str, pathlib.Path], 'name': <class 'str'>, 'path': <class 'pathlib.Path'>, 'type': <enum 'WorkflowType'>}
__dataclass_fields__ = {'additional_options': Field(name='additional_options',type=typing.Dict...appingproxy({}),kw_only=False,_field_type=_FIELD), 'class_name': Field(name='class_name',type=typing.Optional[str...appingproxy({}),kw_only=False,_field_type=_FIELD), 'content_package_version': Field(name='content_package_version',type=typing...appingproxy({}),kw_only=False,_field_type=_FIELD), 'description': Field(name='description',type=<class 'str'>,defa...appingproxy({}),kw_only=False,_field_type=_FIELD), 'dockerfiles': Field(name='dockerfiles',type=typing.Dict[str, p...appingproxy({}),kw_only=False,_field_type=_FIELD), 'name': Field(name='name',type=<class 'str'>,default=<da...appingproxy({}),kw_only=False,_field_type=_FIELD), 'path': Field(name='path',type=<class 'pathlib.Path'>,de...appingproxy({}),kw_only=False,_field_type=_FIELD), 'type': Field(name='type',type=<enum 'WorkflowType'>,def...appingproxy({}),kw_only=False,_field_type=_FIELD)}
__dataclass_params__ = _DataclassParams(init=True,repr=True,eq=True,order=False,unsafe_hash=False,frozen=False)
__hash__ = None
__match_args__ = ('name', 'path', 'type', 'content_package_version', 'class_name', 'dockerfiles', 'description', 'additional_options')
class_name = None
content_package_version = None
dockerfiles = None

 
class WorkflowType(enum.Enum)
    WorkflowType(value, names=None, *, module=None, qualname=None, type=None, start=1)
 
Different types workflows supported.
 
 
Method resolution order:
WorkflowType
enum.Enum
builtins.object

Data and other attributes defined here:
DeploymentYaml = <WorkflowType.DeploymentYaml: 'bright_magenta'>
ExecutionMetaflow = <WorkflowType.ExecutionMetaflow: 'green'>

Data descriptors inherited from enum.Enum:
name
The name of the Enum member.
value
The value of the Enum member.

Readonly properties inherited from enum.EnumMeta:
__members__
Returns a mapping of member name->value.
 
This mapping lists all enum members, including aliases. Note that this
is a read-only view of the internal mapping.

 
Data
        __all__ = ['WorkflowType', 'Workflow']