from abc import abstractmethod
import configparser
from typing import Dict, Optional, Callable
from flambe.compile import MappedRegistrable
from flambe.runnable.utils import DEFAULT_USER_PROVIDER
[docs]class Runnable(MappedRegistrable):
"""Base class for all runnables.
A runnable contains a single run method that needs to
be implemented. It must contain all the logic for
the runnable.
Each runnable has also access to the secrets the user
provides.
Examples of Runnables: Experiment, Cluster
Attributes
----------
config: configparser.ConfigParser
The secrets that the user provides. For example,
'config["AWS"]["ACCESS_KEY"]'
extensions: Dict[str, str]
The extensions used for this runnable.
content: Optional[str]
This attribute will hold the YAML representation
of the Runnable.
user_provider: Callable[[], str]
The logic for specifying the user triggering this
Runnable. If not passed, by default it will pick the computer's
user.
"""
def __init__(self, user_provider: Callable[[], str] = None, **kwargs) -> None:
self.config = configparser.ConfigParser()
self.extensions: Dict[str, str] = {}
self.content: Optional[str] = None
self.user_provider = user_provider or DEFAULT_USER_PROVIDER
[docs] def inject_content(self, content: str) -> None:
"""Inject the original YAML string that was used
to generate this Runnable instance.
Parameters
----------
content: str
The YAML, as a string
"""
self.content = content
[docs] def inject_secrets(self, secrets: str) -> None:
"""Inject the secrets once the Runnable
was created.
Parameters
----------
secrets: str
The filepath to the secrets
"""
self.config.read(secrets)
[docs] def inject_extensions(self, extensions: Dict[str, str]) -> None:
"""Inject extensions to the Runnable
Parameters
----------
extensions: Dict[str, str]
The extensions
"""
self.extensions = extensions
@abstractmethod
[docs] def run(self, **kwargs) -> None:
"""Run the runnable.
Each implementation will implement its
own logic, with the parameters it needs.
"""
raise NotImplementedError()
[docs] def parse(self) -> None:
"""Parse the runnable to determine if it's able to run.
Raises
------
ParsingExperimentError
In case a parsing error is found.
"""
pass