Using better action ids
This commit is contained in:
parent
2487078c8e
commit
33af22443e
|
@ -5,10 +5,13 @@ import click
|
|||
from jeeves.cli.echo import info, error, title, success
|
||||
from jeeves.core.parsers import FlowParser
|
||||
from jeeves.core.executor import Executor
|
||||
from jeeves.core.registry import ActionRegistry
|
||||
|
||||
|
||||
@click.group()
|
||||
def main():
|
||||
# TODO: Check if Jeevesfile in cwd, execute directly
|
||||
ActionRegistry.autodiscover()
|
||||
pass
|
||||
|
||||
|
||||
|
|
|
@ -1,4 +1,12 @@
|
|||
PROVIDED_ACTIONS = [
|
||||
"jeeves.core.actions.shell:ScriptAction",
|
||||
"jeeves.core.actions.docker:DockerAction",
|
||||
from jeeves.core.actions.shell import ScriptAction
|
||||
from jeeves.core.actions.docker import DockerBuildAction, DockerRunAction
|
||||
|
||||
__all__ = [
|
||||
# Shell
|
||||
ScriptAction,
|
||||
# Docker
|
||||
DockerBuildAction,
|
||||
DockerRunAction,
|
||||
]
|
||||
|
||||
PROVIDED_ACTIONS = {action.id: action for action in __all__}
|
||||
|
|
|
@ -6,7 +6,7 @@ from typing import Text
|
|||
import pydantic
|
||||
|
||||
from jeeves.core.objects import Result
|
||||
from .base import Action
|
||||
from jeeves.core.actions.base import Action
|
||||
|
||||
|
||||
class ScriptAction(Action):
|
||||
|
|
|
@ -1,6 +1,7 @@
|
|||
import traceback
|
||||
|
||||
from jeeves.core.objects import Flow, Result, Execution, ExecutionStep
|
||||
from jeeves.core.registry import ActionRegistry
|
||||
|
||||
|
||||
class Executor:
|
||||
|
@ -20,7 +21,8 @@ class Executor:
|
|||
|
||||
def execute_step(self, step: ExecutionStep):
|
||||
try:
|
||||
step.result = step.task.action.execute(workspace=self._execution.workspace)
|
||||
action = ActionRegistry.get_action_cls(step.task.type)(parameters=step.task.parameters)
|
||||
step.result = action.execute(workspace=self._execution.workspace)
|
||||
except Exception as error:
|
||||
# Catch unhandled exceptions, mark the result as unsuccessful
|
||||
# and append the error as output.
|
||||
|
|
|
@ -6,8 +6,6 @@ from dataclasses import field
|
|||
|
||||
import pydantic
|
||||
|
||||
from jeeves.core.registry import ActionRegistry
|
||||
|
||||
|
||||
class BaseObject(pydantic.BaseModel):
|
||||
pass
|
||||
|
@ -23,12 +21,12 @@ class Task(BaseObject):
|
|||
type: Text
|
||||
parameters: Optional[Dict[Any, Any]] = None
|
||||
|
||||
@property
|
||||
def action(self):
|
||||
"""
|
||||
Returns the instanced :any:`jeeves.core.actions.base.Action` defined in ``type`` for this ``Task``.
|
||||
"""
|
||||
return ActionRegistry.get_action_cls(self.type)(parameters=self.parameters)
|
||||
|
||||
class Argument(BaseObject):
|
||||
name: Text
|
||||
default: Any
|
||||
type: Text = "text"
|
||||
required: bool = False
|
||||
|
||||
|
||||
class Flow(BaseObject):
|
||||
|
|
|
@ -27,47 +27,42 @@ class ActionRegistry(metaclass=Singleton):
|
|||
"""Loads all provided actions."""
|
||||
# TODO: Third party plugins
|
||||
registry = cls()
|
||||
for action_namespace in PROVIDED_ACTIONS:
|
||||
registry.register_action(cls.get_action_cls(action_namespace))
|
||||
for action_id, action_cls in PROVIDED_ACTIONS.items():
|
||||
registry.register_action(action_id, action_cls)
|
||||
|
||||
@classmethod
|
||||
def register_action(cls, action_cls):
|
||||
def register_action(cls, action_id, action_cls):
|
||||
registry = cls()
|
||||
# namespace = action_cls.id
|
||||
namespace = f"{action_cls.__module__}:{action_cls.__name__}"
|
||||
|
||||
if namespace in registry.actions:
|
||||
raise cls.ActionNamespaceConflict(
|
||||
f"Namespace {namespace} is already registered"
|
||||
if action_id in registry.actions:
|
||||
raise cls.ActionIDConflict(
|
||||
f"Action ID '{action_id}' is already registered"
|
||||
)
|
||||
|
||||
registry.actions[namespace] = action_cls
|
||||
registry.actions[action_id] = action_cls
|
||||
|
||||
@classmethod
|
||||
def get_action_cls(cls, namespace) -> Type[Action]:
|
||||
"""Returns the class for the provided action namespace"""
|
||||
def get_action_cls(cls, action_id) -> Type[Action]:
|
||||
"""Returns the class for the provided action ID"""
|
||||
# Right now actions are being imported and returned dinamically because it's easier,
|
||||
# but we will need a way of autodiscover all (or register them manually) and
|
||||
# referencing them on a list so the user knows which actions are available.
|
||||
|
||||
modulename, clsname = namespace.split(":")
|
||||
try:
|
||||
module = __import__(f"{modulename}", fromlist=(clsname,), level=0)
|
||||
action_cls = getattr(module, clsname)
|
||||
return action_cls
|
||||
except ModuleNotFoundError as error:
|
||||
raise cls.ActionDoesNotExist(f"Error importing action {namespace}: {error}")
|
||||
return cls.actions[action_id]
|
||||
except IndexError as error:
|
||||
raise cls.ActionDoesNotExist(f"Error importing action {action_id}: {error}")
|
||||
|
||||
@classmethod
|
||||
def get_action(
|
||||
cls, namespace: Text, parameters: Optional[Dict[Any, Any]] = None
|
||||
cls, action_id: Text, parameters: Optional[Dict[Any, Any]] = None
|
||||
) -> Action:
|
||||
"""Returns the instanced action for the provided namespace"""
|
||||
action_cls: Type[Action] = cls.get_action_cls(namespace)
|
||||
"""Returns the instanced action for the provided action_id"""
|
||||
action_cls: Type[Action] = cls.get_action_cls(action_id)
|
||||
return action_cls(parameters=parameters or {})
|
||||
|
||||
class ActionNamespaceConflict(Exception):
|
||||
"""Raised when an action is defined with an already registered namespace"""
|
||||
class ActionIDConflict(Exception):
|
||||
"""Raised when an action is defined with an already registered action id"""
|
||||
|
||||
pass
|
||||
|
||||
|
|
Reference in New Issue