Skip to content

plpipes.action.runner#

lookup(name, parent='') #

Lookup and retrieve an action by its name.

Parameters:

Name Type Description Default
name str

The name of the action to look up.

 required
parent str

An optional parent action name.

 ''

Returns:

Name Type Description
Action

The action instance corresponding to the provided name.

Raises:

Type Description
ValueError

If the action type is not declared or the action file is not found.
Source code in src\plpipes\action\runner.py 
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
def lookup(name, parent=""):
    """
    Lookup and retrieve an action by its name.

    Args:
        name (str): The name of the action to look up.
        parent (str): An optional parent action name.

    Returns:
        Action: The action instance corresponding to the provided name.

    Raises:
        ValueError: If the action type is not declared or the action file is not found.
    """
    name = resolve_action_name(name, parent)
    if name not in _action_cache:
        actions_dir = pathlib.Path(cfg["fs.actions"])
        files = _find_action_files(actions_dir, name)

        cfg_path = "actions." + ".children.".join(name.split("."))
        acfg = cfg.cd(cfg_path)

        for ext in ("yaml", "json"):
            if ext in files:
                acfg.merge_file(files[ext], frame=-1)

        for k, v in files.items():
            acfg.setdefault(f"files.{k}", v)

        action_type = acfg.setdefault("type", _action_type_lookup(files))
        if action_type is None:
            raise ValueError(f"Action {name} has no type declared or action file not found")

        logging.debug(f"action_type: {action_type}")
        _action_cache[name] = _action_class_lookup(action_type)(name, acfg)

    return _action_cache[name]

resolve_action_name(name, parent) #

Resolve the full action name from a relative action name and its parent.

Parameters:

Name Type Description Default
name str

The action name to resolve.

 required
parent str

The parent action name.

 required

Returns:

Name Type Description
str

The resolved full action name.

Raises:

Type Description
ValueError

If the name is a relative name but no parent is provided.
Source code in src\plpipes\action\runner.py 
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
def resolve_action_name(name, parent):
    """
    Resolve the full action name from a relative action name and its parent.

    Args:
        name (str): The action name to resolve.
        parent (str): The parent action name.

    Returns:
        str: The resolved full action name.

    Raises:
        ValueError: If the name is a relative name but no parent is provided.
    """
    if name.startswith("."):
        if parent:
            return f"{parent}{name}"
        else:
            raise ValueError("Can't resolve relative action name without a parent")
    return name

run(name) #

Execute the action corresponding to the provided name.

Parameters:

Name Type Description Default
name str

The name of the action to run.

 required
Source code in src\plpipes\action\runner.py 
100
101
102
103
104
105
106
107
def run(name):
    """
    Execute the action corresponding to the provided name.

    Args:
        name (str): The name of the action to run.
    """
    lookup(name).run()