registry

class pcs.registry.InMemoryRegistry(input_dict: Optional[Dict] = None, base_dir: Optional[str] = None)

A mutable in-memory registry.

__init__(input_dict: Optional[Dict] = None, base_dir: Optional[str] = None)

add_component_spec(component_spec: Mapping[str, Mapping[str, str]]) → None

Adds a component spec to this registry. This does not add any Registry Objects to this registry.

To register a component, use the higher the level function Component.register(registry) or Registry.add_component().

Parameters

component_spec – The ComponentSpec to register.

add_repo_spec(repo_spec: Union[Mapping[str, Mapping[str, str]], Mapping[str, str]]) → None

add_run_command_spec(run_command_spec: Mapping) → None

add_run_spec(run_spec: Mapping) → None

get_component_specs(filter_by_name: Optional[str] = None, filter_by_version: Optional[str] = None) → Mapping[str, Mapping[str, str]]

Return dictionary of component specs in this Registry, optionally filtered by name and/or version; or None if none are found.

Each Component Spec is itself a dict mapping ComponentIdentifier to a dict of properties that define the Component.

Optionally, filter the list to match all filter strings provided. Filters can be provided on name, version, or both.

If this registry contains zero component specs that match the filter criteria (if any), then an empty dictionary is returned.

If filter_by_name and filter_by_version are provided, then 0 or 1 components will be returned.

Parameters

• filter_by_name – return only components with this name.

• filter_by_version – return only components with this version.

• include_id_in_contents – add name and version fields to innermost dict of the ComponentSpec Dict. This denormalizes the spec by duplicating the name and version which are already included via the ComponentIdentifier Dict key.

Returns

A dictionary of components in this registry, optionally filtered by name, version, or both. If no matching components are found, an empty dictionary is returned.

get_registries() → Sequence[pcs.registry.Registry]

get_repo_spec(repo_id: str, flatten: bool = False, error_if_not_found: bool = True) → Optional[Union[Mapping[str, Mapping[str, str]], Mapping[str, str]]]

get_run_command_spec(run_command_id: str, flatten: bool = False, error_if_not_found: bool = True) → Optional[Mapping]

get_run_spec(run_id: str, flatten: bool = False, error_if_not_found: bool = True) → Optional[Mapping]

to_dict() → Dict

class pcs.registry.Registry(base_dir: Optional[str] = None)

__init__(base_dir: Optional[str] = None)

add_component(component: Component, recurse: bool = True, force: bool = False) → None

abstract add_component_spec(component_spec: Mapping[str, Mapping[str, str]]) → None

Adds a component spec to this registry. This does not add any Registry Objects to this registry.

To register a component, use the higher the level function Component.register(registry) or Registry.add_component().

Parameters

component_spec – The ComponentSpec to register.

abstract add_repo_spec(repo_spec: Union[Mapping[str, Mapping[str, str]], Mapping[str, str]]) → None

abstract add_run_command_spec(run_command_spec: Mapping) → None

abstract add_run_spec(run_spec: Mapping) → None

classmethod from_default()

static from_dict(input_dict: Dict) → pcs.registry.Registry

classmethod from_file_in_repo(repo: Repo, file_path: str, version: str, format: str = 'yaml') → Registry

Read in a registry file from an repo.

Parameters

• repo – Repo to load registry file from.

• file_path – Path within Repo that registry is located, relative to the repo root.

• format – Optionally specify the format of the registry file.

Returns

a new Registry object.

classmethod from_repo(repo: Repo)

Get a registry from a Repo. If the Repo has a default registry file, use that, if not infer specs by inspecting the contents of the repo.

classmethod from_repo_inferred(repo: Repo, version: str = None, py_file_suffixes: Tuple[str] = ('.py', '.python'), requirements_file: str = 'requirements.txt')

static from_yaml(file_path: str) → pcs.registry.Registry

get_component_spec(name: str, version: Optional[str] = None, flatten: bool = False, error_if_not_found: bool = True) → Optional[Union[Mapping[str, Mapping[str, str]], Mapping[str, str]]]

Returns the component spec with name and version, if it exists, or raise an Error if it does not. A component’s name and version are defined as its identifier’s name and version.

Registries are not allowed to contain multiple Components with the same identifier. The Registry abstract base class does not enforce that all Components have a version (version can be None) though some sub-classes, such as web service backed registries, may choose to enforce that constraint.

When version is unspecified or None, this function assumes that a Component c exists where c.name == name and c.version is None, and throws an error otherwise.

Subclasses of Registry may choose to provide their own (more elaborate) semantics for “default components”. E.g., since WebRegistry does not allow non-versioned components, it defines its own concept of a default component by maintaining a separate map from component name to a specific version of the component, and it allows that mapping to be updated by users.

Parameters

• name – The name of the component to fetch.

• version – Optional version of the component to fetch.

• flatten – If True, flatten the outermost 2 layers of nested dicts into a single dict. In an unflattened component spec, the outermost dict is from identifier (which is a string in the format of name[==version]) Component component properties (class_name, repo, etc.). In a flattened Component spec, the name and version are included in the same dictionary as the class_name, repo, dependencies, etc.

• error_if_not_found – Set to False to return an empty dict in the case that a matching component is not found in this registry.

Returns

a ComponentSpec (i.e. a dict) matching the filter criteria provided, else throw an error.

get_component_spec_by_id(identifier: Union[pcs.identifiers.ComponentIdentifier, str], flatten: bool = False) → Optional[Union[Mapping[str, Mapping[str, str]], Mapping[str, str]]]

abstract get_component_specs(filter_by_name: Optional[str] = None, filter_by_version: Optional[str] = None) → Mapping[str, Mapping[str, str]]

Return dictionary of component specs in this Registry, optionally filtered by name and/or version; or None if none are found.

Each Component Spec is itself a dict mapping ComponentIdentifier to a dict of properties that define the Component.

Optionally, filter the list to match all filter strings provided. Filters can be provided on name, version, or both.

If this registry contains zero component specs that match the filter criteria (if any), then an empty dictionary is returned.

If filter_by_name and filter_by_version are provided, then 0 or 1 components will be returned.

Parameters

• filter_by_name – return only components with this name.

• filter_by_version – return only components with this version.

• include_id_in_contents – add name and version fields to innermost dict of the ComponentSpec Dict. This denormalizes the spec by duplicating the name and version which are already included via the ComponentIdentifier Dict key.

Returns

A dictionary of components in this registry, optionally filtered by name, version, or both. If no matching components are found, an empty dictionary is returned.

abstract get_registries() → Sequence

abstract get_repo_spec(repo_id: str, flatten: bool = False, error_if_not_found: bool = True) → Optional[Union[Mapping[str, Mapping[str, str]], Mapping[str, str]]]

get_run_command_spec(run_command_id: str, flatten: bool = False, error_if_not_found: bool = True) → Optional[Mapping]

abstract get_run_spec(run_id: str, flatten: bool = False, error_if_not_found: bool = True) → Optional[Mapping]

get_specs_transitively_by_id(identifier: Union[pcs.identifiers.ComponentIdentifier, str], flatten: bool = True)

has_component_by_id(identifier: pcs.identifiers.ComponentIdentifier) → bool

has_component_by_name(name: str, version: Optional[str] = None) → bool

abstract to_dict() → Dict

to_yaml(filename: str) → None

class pcs.registry.WebRegistry(root_api_url: str, base_dir: Optional[str] = None)

A web-server backed Registry.

__init__(root_api_url: str, base_dir: Optional[str] = None)

add_component_spec(component_spec: Mapping[str, Mapping[str, str]]) → None

Adds a component spec to this registry. This does not add any Registry Objects to this registry.

To register a component, use the higher the level function Component.register(registry) or Registry.add_component().

Parameters

component_spec – The ComponentSpec to register.

add_repo_spec(repo_spec: Union[Mapping[str, Mapping[str, str]], Mapping[str, str]]) → None

add_run_artifacts(run_id: int, run_artifact_paths: Sequence[str]) → Sequence

add_run_command_spec(run_command_spec: Mapping) → None

add_run_spec(run_spec: Mapping) → Sequence

get_component_specs(filter_by_name: Optional[str] = None, filter_by_version: Optional[str] = None) → Mapping[str, Mapping[str, str]]

Return dictionary of component specs in this Registry, optionally filtered by name and/or version; or None if none are found.

Each Component Spec is itself a dict mapping ComponentIdentifier to a dict of properties that define the Component.

Optionally, filter the list to match all filter strings provided. Filters can be provided on name, version, or both.

If this registry contains zero component specs that match the filter criteria (if any), then an empty dictionary is returned.

If filter_by_name and filter_by_version are provided, then 0 or 1 components will be returned.

Parameters

• filter_by_name – return only components with this name.

• filter_by_version – return only components with this version.

• include_id_in_contents – add name and version fields to innermost dict of the ComponentSpec Dict. This denormalizes the spec by duplicating the name and version which are already included via the ComponentIdentifier Dict key.

Returns

A dictionary of components in this registry, optionally filtered by name, version, or both. If no matching components are found, an empty dictionary is returned.

get_default_component(name: str)

get_registries() → Sequence

get_repo_spec(repo_id: str, flatten: bool = False, error_if_not_found: bool = True) → Optional[Union[Mapping[str, Mapping[str, str]], Mapping[str, str]]]

get_run(run_id: str) → Run

get_run_command_spec(run_command_id: str, flatten: bool = False, error_if_not_found: bool = True) → Mapping

get_run_spec(run_id: str, flatten: bool = False, error_if_not_found: bool = True) → Mapping

to_dict() → Dict