16651837049

Committed 31 Jul 2025 02:28PM UTC coverage: 84.483%. First build

Build # 16651837049

Build Type

push

github

Committed by

teobouvard

Commit Message

Add initial project structure

Run Details

49 of 58 new or added lines in 2 files covered. (84.48%)

49 of 58 relevant lines covered (84.48%)

0.84 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

80.43

/src/commandfile_argo/generator.py

import json
import logging
import os
from argparse import ArgumentParser
from pathlib import Path

from commandfile.io import write_cmdfile_yaml
from commandfile.model import Commandfile, Filelist, Parameter

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s.%(msecs)03d %(levelname)s %(message)s",
    datefmt="%Y-%m-%dT%H:%M:%S",
)
logger = logging.getLogger(__name__)


def get_args():
    parser = ArgumentParser(
        description="Generate a commandfile from Argo workflow environment."
    )
    parser.add_argument("output", type=Path, help="Path to the output commandfile.")
    return parser.parse_args()


def generate_commandfile(output_file: Path):
    try:
        template_contents = os.environ["ARGO_TEMPLATE"]
    except KeyError:
        raise ValueError("Environment variable 'ARGO_TEMPLATE' is not set.")

    try:
        parameters_contents = os.environ["INPUTS_PARAMETERS"]
    except KeyError:
        raise ValueError("Environment variable 'INPUTS_PARAMETERS' is not set.")

    logger.info("ARGO_TEMPLATE: %s", template_contents)
    logger.info("INPUTS_PARAMETERS: %s", parameters_contents)

    argo_template = json.loads(template_contents)
    argo_parameters = json.loads(parameters_contents)
    params, input_files = transform_complex_parameters(argo_parameters)

    commandfile = Commandfile(
        header={},
        parameters=params,
        inputs=[
            Filelist(key=a["name"], files=[a["path"]])
            for a in argo_template["inputs"].get("artifacts", [])
        ]
        + input_files,
        outputs=[
            Filelist(key=a["name"], files=[a["path"]])
            for a in argo_template["outputs"].get("artifacts", [])
        ],
    )
    write_cmdfile_yaml(commandfile, output_file)
    logger.info(
        "Written command file to %s:\n%s",
        output_file,
        commandfile.model_dump_json(),
    )


def transform_complex_parameters(
    params: list,
) -> tuple[list[Parameter], list[Filelist]]:
    parameters = []
    input_files = []
    for param in params:
        try:
            # Parameters containing JSON arrays are implicitly transformed to input
            # filelists.
            content = json.loads(param["value"])
            if isinstance(content, list) and all(
                isinstance(item, str) for item in content
            ):
                input_files.append(Filelist(key=param["name"], files=content))
                continue
        except json.JSONDecodeError:
            pass

        # For other parameters, we keep them as-is.
        parameters.append(Parameter(key=param["name"], value=param["value"]))

    return parameters, input_files


def main():
    args = get_args()
    generate_commandfile(args.output)

1	import json	1✔
2	import logging	1✔
3	import os	1✔
4	from argparse import ArgumentParser	1✔
5	from pathlib import Path	1✔
6
7	from commandfile.io import write_cmdfile_yaml	1✔
8	from commandfile.model import Commandfile, Filelist, Parameter	1✔
9
10	logging.basicConfig(	1✔
11	level=logging.INFO,
12	format="%(asctime)s.%(msecs)03d %(levelname)s %(message)s",
13	datefmt="%Y-%m-%dT%H:%M:%S",
14	)
15	logger = logging.getLogger(__name__)	1✔
16
17
18	def get_args():	1✔
NEW 19	parser = ArgumentParser(	×
20	description="Generate a commandfile from Argo workflow environment."
21	)
NEW 22	parser.add_argument("output", type=Path, help="Path to the output commandfile.")	×
NEW 23	return parser.parse_args()	×
24
25
26	def generate_commandfile(output_file: Path):	1✔
27	try:	1✔
28	template_contents = os.environ["ARGO_TEMPLATE"]	1✔
NEW 29	except KeyError:	×
NEW 30	raise ValueError("Environment variable 'ARGO_TEMPLATE' is not set.")	×
31
32	try:	1✔
33	parameters_contents = os.environ["INPUTS_PARAMETERS"]	1✔
NEW 34	except KeyError:	×
NEW 35	raise ValueError("Environment variable 'INPUTS_PARAMETERS' is not set.")	×
36
37	logger.info("ARGO_TEMPLATE: %s", template_contents)	1✔
38	logger.info("INPUTS_PARAMETERS: %s", parameters_contents)	1✔
39
40	argo_template = json.loads(template_contents)	1✔
41	argo_parameters = json.loads(parameters_contents)	1✔
42	params, input_files = transform_complex_parameters(argo_parameters)	1✔
43
44	commandfile = Commandfile(	1✔
45	header={},
46	parameters=params,
47	inputs=[
48	Filelist(key=a["name"], files=[a["path"]])
49	for a in argo_template["inputs"].get("artifacts", [])
50	]
51	+ input_files,
52	outputs=[
53	Filelist(key=a["name"], files=[a["path"]])
54	for a in argo_template["outputs"].get("artifacts", [])
55	],
56	)
57	write_cmdfile_yaml(commandfile, output_file)	1✔
58	logger.info(	1✔
59	"Written command file to %s:\n%s",
60	output_file,
61	commandfile.model_dump_json(),
62	)
63
64
65	def transform_complex_parameters(	1✔
66	params: list,
67	) -> tuple[list[Parameter], list[Filelist]]:
68	parameters = []	1✔
69	input_files = []	1✔
70	for param in params:	1✔
71	try:	1✔
72	# Parameters containing JSON arrays are implicitly transformed to input
73	# filelists.
74	content = json.loads(param["value"])	1✔
75	if isinstance(content, list) and all(	1✔
76	isinstance(item, str) for item in content
77	):
78	input_files.append(Filelist(key=param["name"], files=content))	1✔
79	continue	1✔
80	except json.JSONDecodeError:	1✔
81	pass	1✔
82
83	# For other parameters, we keep them as-is.
84	parameters.append(Parameter(key=param["name"], value=param["value"]))	1✔
85
86	return parameters, input_files	1✔
87
88
89	def main():	1✔
NEW 90	args = get_args()	×
NEW 91	generate_commandfile(args.output)	×

lisa-sgs / commandfile-argo / 16651837049

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous