component

class pcs.component.Component(repo: pcs.repo.Repo, identifier: pcs.identifiers.ComponentIdentifier, file_path: str, class_name: Optional[str] = None, instantiate: bool = False, requirements_path: Optional[str] = None, dependencies: Optional[Dict] = None, use_venv: bool = True, dunder_name: Optional[str] = None)

A Component is an object manager. Objects can be Python Modules, Python Classes, or Python Class Instances. The Component abstraction provides a standard programmatic mechanism for managing dependencies between these objects, reproducibly creating/initializing them and executing their methods. You can think of methods on a managed object as “managed methods” which we call “Entry Points”. We call the execution of an Entry Point a “Run”. Components provide reproducibility by automatically tracking (i.e., logging) all of the parts that make up a Run, including: (1) the code of the object being run (i.e., the Component and its Entry Point), (2) the full DAG of other objects it depends on (i.e., DAG of other Components), (3) the set of arguments (literally an ArgumentSet) used during initialization of the managed object and all objects it transitively depends on, and (4) the arguments passed to the Entry Point being run.

__init__(repo: pcs.repo.Repo, identifier: pcs.identifiers.ComponentIdentifier, file_path: str, class_name: Optional[str] = None, instantiate: bool = False, requirements_path: Optional[str] = None, dependencies: Optional[Dict] = None, use_venv: bool = True, dunder_name: Optional[str] = None)
Parameters

  • repo – Repo where this component’s module file can be found. The file_path argument is relative to the root this Repo.

  • identifier – Used to identify the Component.

  • file_path – Path to Python module file this Component manages.

  • class_name – Optionally, the name of the class that is being managed. If none provided, then by default this component is a managed Python Module.

  • instantiate – Optional. If True, this Component is a managed Python Class Instance, class_name must also be passed, and get_object() returns an instance of the class with name class_name. If False and class_name is provided, then this Component is a managed Python Class and get_object() returns a Python Class object specified by class_name. If False and class_name is not provided, this Component is a managed Python Module and get_object() returns a Python Module object.

  • requirements_path – Optional path to a pip installable file.

  • dependencies – List of other components that self depends on.

  • use_venv – Whether to create a VM when setting up the object this component manages.

  • dunder_name – Name used for the pointer to this Component on any managed objects created by this Component.

add_dependency(component: pcs.component.Component, attribute_name: Optional[str] = None) → None
call_function_with_arg_set(instance: Any, function_name: str, arg_set: pcs.argument_set.ArgumentSet) → Any
dependency_list(include_root: bool = True, include_parents: bool = False) → Sequence[pcs.component.Component]

Return a normalized (i.e. flat) Sequence containing all transitive dependencies of this component and (optionally) this component.

Parameters

  • include_root – Whether to include root component in the list. If True, self is included in the list returned.

  • include_parents – If True, then recursively include all parents of this component (and their parents, etc). A parent of this Component is a Component which depends on this Component. Ultimately, if True, all Components in the DAG will be returned.

Returns

a list containing all all of the transitive dependencies of this component (optionally including the root component).

classmethod from_class(class_obj: Type[pcs.component.T], repo: Optional[pcs.repo.Repo] = None, identifier: Optional[str] = None, instantiate: bool = False, use_venv: bool = True, dunder_name: Optional[str] = None) → pcs.component.Component
classmethod from_default_registry(name: str, version: Optional[str] = None, use_venv: bool = True) → pcs.component.Component
classmethod from_github_registry(github_url: str, name: str, version: Optional[str] = None, use_venv: bool = True) → pcs.component.Component

This method gets a Component from a registry file found on GitHub. If the registry file contains a LocalRepo, this method automatically translates that LocalRepo into a GitHubRepo. Pass use_venv=False if you want to import and run the Component in your existing Python environment.

The github_url argument can be found by navigating to the registry file on the GitHub web UI. It should look like the following:

https://github.com/<project>/<repo>/{blob,raw}/<branch>/<path>
classmethod from_registry(registry: pcs.registry.Registry, name: str, version: Optional[str] = None, use_venv: bool = True) → pcs.component.Component

Returns a Component Object from the provided registry, including its full dependency tree of other Component Objects. If no Registry is provided, use the default registry.

classmethod from_registry_file(yaml_file: str, name: str, version: Optional[str] = None, use_venv: bool = True) → pcs.component.Component
classmethod from_repo(repo: pcs.repo.Repo, identifier: str, file_path: str, class_name: Optional[str] = None, instantiate: bool = False, requirements_path: Optional[str] = None, use_venv: bool = True, dunder_name: Optional[str] = None) → pcs.component.Component
get_default_entry_point()
get_object(arg_set: Optional[pcs.argument_set.ArgumentSet] = None) → pcs.component.T
get_status_tree(parent_tree: Optional[rich.tree.Tree] = None) → rich.tree.Tree
property name: str
print_status_tree() → None
run(entry_point: str, **kwargs)

Run an entry point with provided arguments. If you need to specify arguments to the init function of the managed object or any of its dependency components, use :py:func:run_with_arg_set:.

Parameters

  • entry_point – name of function to call on manage object.

  • kwargs – keyword-only args to pass through to managed object function called entry-point.

Returns

the return value of the entry point called.

run_with_arg_set(entry_point: str, args: Optional[Union[pcs.argument_set.ArgumentSet, Dict]] = None, publish_to: Optional[pcs.registry.Registry] = None, log_return_value: bool = True, return_value_log_format: str = 'yaml') → pcs.component_run.ComponentRun

Run the specified entry point a new instance of this Component’s managed object given the specified args, log the results and return the Run object.

Parameters

  • entry_point – Name of a function to be called on a new instance of this component’s managed object.

  • args – A :py:func:agentos.argument_set.ArgumentSet: or ArgumentSet-like dict containing the entry-point arguments, and/or arguments to be passed to the __init__() functions of this component’s dependents during managed object initialization.

  • publish_to – Optionally, publish the resulting Run object to the provided registry.

  • log_return_value – If True, log the return value of the entry point being run.

  • return_value_log_format – Specify which format to use when serializing the return value. Only used if log_return_value is True.

to_frozen_registry(force: bool = False) → pcs.registry.Registry
to_registry(registry: Optional[pcs.registry.Registry] = None, recurse: bool = True, force: bool = False) → pcs.registry.Registry

Returns a registry containing specs for this component, all of its transitive dependents, and the repos of all of them. Throws an exception if any of them already exist in the Registry that are different unless force is set to True.

Parameters

  • registry – Optionally, add the component spec for this component and each of its transitive dependencies (which are themselves components) to the specified registry.

  • recurse – If True, check that all transitive dependencies exist in the registry already, and if they don’t, then add them. This includes dependencies on other Components as well as dependencies a repo. If they do, ensure that they are equal to this component’s dependencies (unless force is specified).

  • force – Optionally, if a component with the same identifier already exists and is different than the current one, attempt to overwrite the registered one with this one.

to_spec(flatten: bool = False) → Union[Mapping[str, Mapping[str, str]], Mapping[str, str]]
property version