Validate schema arguments against required/forbidden ones.
| PARAMETER |
DESCRIPTION |
task_type
|
TYPE:
Literal['parallel', 'non_parallel', 'compound']
|
executable_kind
|
The parallel/non_parallel part of the task.
TYPE:
Literal['parallel', 'non_parallel']
|
schema
|
TYPE:
dict[str, Any]
|
Source code in fractal_task_tools/_task_arguments.py
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
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75 | def validate_arguments(
*,
task_type: Literal["parallel", "non_parallel", "compound"],
executable_kind: Literal["parallel", "non_parallel"],
schema: dict[str, Any],
) -> None:
"""
Validate schema arguments against required/forbidden ones.
Arguments:
task_type:
executable_kind: The `parallel`/`non_parallel` part of the task.
schema:
"""
key = (task_type, executable_kind)
if not (key in REQUIRED_ARGUMENTS and key in FORBIDDEN_ARGUMENTS):
logging.error(f"Invalid {task_type=}, {executable_kind=}.")
raise ValueError(f"Invalid {task_type=}, {executable_kind=}.")
required_args = REQUIRED_ARGUMENTS[key]
forbidden_args = FORBIDDEN_ARGUMENTS[key]
schema_properties = set(schema["properties"].keys())
logging.info(
f"[validate_arguments] Task has arguments: {schema_properties}"
)
logging.info(f"[validate_arguments] Required arguments: {required_args}")
logging.info(f"[validate_arguments] Forbidden arguments: {forbidden_args}")
missing_required_arguments = {
arg for arg in required_args if arg not in schema_properties
}
if missing_required_arguments:
error_msg = (
"[validate_arguments] Required arguments "
f"{missing_required_arguments} are missing."
)
logging.error(error_msg)
raise ValueError(error_msg)
present_forbidden_args = forbidden_args.intersection(schema_properties)
if present_forbidden_args:
error_msg = (
"[validate_arguments] Forbidden arguments "
f"{present_forbidden_args} are present."
)
logging.error(error_msg)
raise ValueError(error_msg)
|