Skip to content

_task_arguments

validate_arguments(*, task_type, executable_kind, schema)

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 src/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)