Skip to content

_signature_constraints

_extract_function(module_relative_path, function_name, package_name, verbose=False)

Extract function from a module with the same name.

PARAMETER DESCRIPTION
package_name

Example fractal_tasks_core.

TYPE: str

module_relative_path

Example tasks/create_ome_zarr.py.

TYPE: str

function_name

Example create_ome_zarr.

TYPE: str

verbose

TYPE: bool DEFAULT: False

Source code in fractal_task_tools/_signature_constraints.py
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
def _extract_function(
    module_relative_path: str,
    function_name: str,
    package_name: str,
    verbose: bool = False,
) -> callable:
    """
    Extract function from a module with the same name.

    Args:
        package_name: Example `fractal_tasks_core`.
        module_relative_path: Example `tasks/create_ome_zarr.py`.
        function_name: Example `create_ome_zarr`.
        verbose:
    """
    if not module_relative_path.endswith(".py"):
        raise ValueError(f"{module_relative_path=} must end with '.py'")
    module_relative_path_no_py = str(
        Path(module_relative_path).with_suffix("")
    )
    module_relative_path_dots = module_relative_path_no_py.replace("/", ".")
    if verbose:
        logging.info(
            f"Now calling `import_module` for "
            f"{package_name}.{module_relative_path_dots}"
        )
    imported_module = import_module(
        f"{package_name}.{module_relative_path_dots}"
    )
    if verbose:
        logging.info(
            f"Now getting attribute {function_name} from "
            f"imported module {imported_module}."
        )
    task_function = getattr(imported_module, function_name)
    return task_function

_validate_function_signature(function)

Validate the function signature.

Implement a set of checks for type hints that do not play well with the creation of JSON Schema, see https://github.com/fractal-analytics-platform/fractal-tasks-core/issues/399.

PARAMETER DESCRIPTION
function

TBD

TYPE: callable

Source code in fractal_task_tools/_signature_constraints.py
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
def _validate_function_signature(function: callable):
    """
    Validate the function signature.

    Implement a set of checks for type hints that do not play well with the
    creation of JSON Schema, see
    https://github.com/fractal-analytics-platform/fractal-tasks-core/issues/399.

    Args:
        function: TBD
    """
    sig = signature(function)
    for param in sig.parameters.values():
        # CASE 1: Check that name is not forbidden
        if param.name in FORBIDDEN_PARAM_NAMES:
            raise ValueError(
                f"Function {function} has argument with forbidden "
                f"name '{param.name}'"
            )

        annotation_is_union = is_union(param.annotation)
        annotation_str = str(param.annotation)
        annotation_has_default = (param.default is not None) and (
            param.default != inspect._empty
        )

        if annotation_is_union:
            if annotation_str.count("|") > 1 or annotation_str.count(",") > 1:
                raise ValueError(
                    "Only unions of two elements are supported, but parameter "
                    f"'{param.name}' has type hint '{annotation_str}'."
                )
            elif (
                "None" not in annotation_str
                and "Optional[" not in annotation_str
            ):
                raise ValueError(
                    "One union element must be None, but parameter "
                    f"'{param.name}' has type hint '{annotation_str}'."
                )
            elif annotation_has_default:
                raise ValueError(
                    "Non-None default not supported, but parameter "
                    f"'{param.name}' has type hint '{annotation_str}' "
                    f"and default {param.default}."
                )

    logging.info("[_validate_function_signature] END")
    return sig