Skip to content

autora.workflow.__main__

_load_function(fully_qualified_function_name)

Load a function by its fully qualified name, module.function_name

Source code in autora/workflow/__main__.py 
70
71
72
73
74
75
76
77
def _load_function(fully_qualified_function_name: str):
    """Load a function by its fully qualified name, `module.function_name`"""
    _logger.debug(f"_load_function: Loading function {fully_qualified_function_name}")
    module_name, function_name = fully_qualified_function_name.rsplit(".", 1)
    module = importlib.import_module(module_name)
    function = getattr(module, function_name)
    _logger.debug(f"_load_function: Loaded function {function} from {module}")
    return function

main(fully_qualified_function_name, in_path=None, in_loader=default_serializer, out_path=None, out_dumper=default_serializer, verbose=False, debug=False)

Run an arbitrary function on an optional input State object and save the output.

Source code in autora/workflow/__main__.py 
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
def main(
    fully_qualified_function_name: Annotated[
        str,
        typer.Argument(
            help="Fully qualified name of the function to load, like `module.function`"
        ),
    ],
    in_path: Annotated[
        Optional[pathlib.Path],
        typer.Option(help="Path to a file with the initial state"),
    ] = None,
    in_loader: Annotated[
        SupportedSerializer,
        typer.Option(
            help="(de)serializer to load the data",
        ),
    ] = default_serializer,
    out_path: Annotated[
        Optional[pathlib.Path],
        typer.Option(help="Path to output the final state"),
    ] = None,
    out_dumper: Annotated[
        SupportedSerializer,
        typer.Option(
            help="serializer to save the data",
        ),
    ] = default_serializer,
    verbose: Annotated[bool, typer.Option(help="Turns on info logging level.")] = False,
    debug: Annotated[bool, typer.Option(help="Turns on debug logging level.")] = False,
):
    """Run an arbitrary function on an optional input State object and save the output."""
    _configure_logger(debug, verbose)
    starting_state = load_state(in_path, in_loader)
    _logger.info(f"Starting State: {starting_state}")
    function = _load_function(fully_qualified_function_name)
    ending_state = function(starting_state)
    _logger.info(f"Ending State: {ending_state}")
    dump_state(ending_state, out_path, out_dumper)

    return