Skip to content

autora.workflow.__main__

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