_args_schemas

_create_schema_for_function(function)

Create JSON Schema for a given function.

PARAMETER	DESCRIPTION
function	TYPE: callable

Source code in src/fractal_task_tools/_args_schemas.py

def _create_schema_for_function(function: callable) -> JSONdictType:
    """
    Create JSON Schema for a given function.

    Args:
        function:
    """

    from pydantic import ConfigDict
    from pydantic.experimental.arguments_schema import generate_arguments_schema
    from pydantic.fields import ComputedFieldInfo
    from pydantic.fields import FieldInfo

    # NOTE: v2.12.0 modified the generated field titles. The function
    # `make_title` restores the `<2.12.0` behavior
    def make_title(name: str, info: FieldInfo | ComputedFieldInfo):
        return name.title().replace("_", " ").strip()

    core_schema = generate_arguments_schema(
        function,
        schema_type="arguments",
        config=ConfigDict(field_title_generator=make_title),
    )

    gen_json_schema = CustomGenerateJsonSchema()
    json_schema = gen_json_schema.generate(core_schema, mode="validation")
    return json_schema

_remove_attributes_from_descriptions(old_schema)

Keep only the description part of the docstrings.

E.g. go from

'Custom class for Omero-channel window, based on OME-NGFF v0.4.\n'
'\n'
'Attributes:\n'
'min: Do not change. It will be set to `0` by default.\n'
'max: Do not change. It will be set according to bitdepth of the images\n'
'    by default (e.g. 65535 for 16 bit images).\n'
'start: Lower-bound rescaling value for visualization.\n'
'end: Upper-bound rescaling value for visualization.'

to 'Custom class for Omero-channel window, based on OME-NGFF v0.4.\n'.

PARAMETER	DESCRIPTION
old_schema	The JSON schema to be modified. TYPE: JSONdictType

Source code in src/fractal_task_tools/_args_schemas.py

def _remove_attributes_from_descriptions(old_schema: JSONdictType) -> JSONdictType:
    """
    Keep only the description part of the docstrings.

    E.g. go from
    ```
    'Custom class for Omero-channel window, based on OME-NGFF v0.4.\\n'
    '\\n'
    'Attributes:\\n'
    'min: Do not change. It will be set to `0` by default.\\n'
    'max: Do not change. It will be set according to bitdepth of the images\\n'
    '    by default (e.g. 65535 for 16 bit images).\\n'
    'start: Lower-bound rescaling value for visualization.\\n'
    'end: Upper-bound rescaling value for visualization.'
    ```
    to `'Custom class for Omero-channel window, based on OME-NGFF v0.4.\\n'`.

    Args:
        old_schema: The JSON schema to be modified.
    """
    new_schema = old_schema.copy()
    if "$defs" in new_schema:
        for name, definition in new_schema["$defs"].items():
            if "description" in definition.keys():
                parsed_docstring = docparse(definition["description"])
                new_schema["$defs"][name]["description"] = (
                    parsed_docstring.short_description
                )
            elif "title" in definition.keys():
                title = definition["title"]
                new_schema["$defs"][name]["description"] = (
                    f"Missing description for {title}."
                )
            else:
                new_schema["$defs"][name]["description"] = "Missing description"
    logging.info("[_remove_attributes_from_descriptions] END")
    return new_schema

_remove_top_level_single_element_allof(schema)

Transform "allOf": [{"$ref": X}] into "$ref": X

PARAMETER	DESCRIPTION
schema	TYPE: JSONdictType

Source code in src/fractal_task_tools/_args_schemas.py

def _remove_top_level_single_element_allof(schema: JSONdictType) -> JSONdictType:
    """
    Transform `"allOf": [{"$ref": X}]` into `"$ref": X`

    Args:
        schema:
    """
    old_schema = deepcopy(schema)
    for arg_name, old_arg_schema in old_schema["properties"].items():
        if (
            "allOf" in old_arg_schema.keys()
            and isinstance(old_arg_schema["allOf"], list)
            and len(old_arg_schema["allOf"]) == 1
            and isinstance(old_arg_schema["allOf"][0], dict)
            and list(old_arg_schema["allOf"][0].keys()) == ["$ref"]
            and "$ref" not in old_arg_schema.keys()
        ):
            new_arg_schema: dict[str, JSONType] = deepcopy(old_arg_schema)
            key_value = new_arg_schema.pop("allOf")[0]
            new_arg_schema.update(key_value)
            schema["properties"][arg_name] = new_arg_schema
            logging.debug(f"Replaced single-item allOf with {key_value} ")
    return schema

create_schema_for_single_task(executable, package=None, task_function=None, verbose=False)

Main function to create a JSON Schema of task arguments

This function can be used in two ways:

1. task_function argument is None, package is set, and executable is a path relative to that package.
2. task_function argument is provided, executable is an absolute path to the function module, and package is `None. This is useful for testing.

PARAMETER	DESCRIPTION
executable	TYPE: str
package	TYPE: Optional[str] DEFAULT: None
task_function	TYPE: Optional[callable] DEFAULT: None
verbose	TYPE: bool DEFAULT: False

Source code in src/fractal_task_tools/_args_schemas.py

def create_schema_for_single_task(
    executable: str,
    package: Optional[str] = None,
    task_function: Optional[callable] = None,
    verbose: bool = False,
) -> JSONdictType:
    """
    Main function to create a JSON Schema of task arguments

    This function can be used in two ways:

    1. `task_function` argument is `None`, `package` is set, and `executable`
        is a path relative to that package.
    2. `task_function` argument is provided, `executable` is an absolute path
        to the function module, and `package` is `None. This is useful for
        testing.

    Args:
        executable:
        package:
        task_function:
        verbose:
    """

    logging.info("[create_schema_for_single_task] START")
    if task_function is None:
        usage = "1"
        # Usage 1 (standard)
        if package is None:
            raise ValueError(
                "Cannot call `create_schema_for_single_task with "
                f"{task_function=} and {package=}. Exit."
            )
        if os.path.isabs(executable):
            raise ValueError(
                "Cannot call `create_schema_for_single_task with "
                f"{task_function=} and absolute {executable=}. Exit."
            )
    else:
        usage = "2"
        # Usage 2 (testing)
        if package is not None:
            raise ValueError(
                "Cannot call `create_schema_for_single_task with "
                f"{task_function=} and non-None {package=}. Exit."
            )
        if not os.path.isabs(executable):
            raise ValueError(
                "Cannot call `create_schema_for_single_task with "
                f"{task_function=} and non-absolute {executable=}. Exit."
            )

    # Extract function from module
    if usage == "1":
        # Extract the function name (for the moment we assume the function has
        # the same name as the module)
        function_name = Path(executable).with_suffix("").name
        # Extract the function object
        task_function = _extract_function(
            package_name=package,
            module_relative_path=executable,
            function_name=function_name,
            verbose=verbose,
        )
    else:
        # The function object is already available, extract its name
        function_name = task_function.__name__

    if verbose:
        logging.info(f"[create_schema_for_single_task] {function_name=}")
        logging.info(f"[create_schema_for_single_task] {task_function=}")

    # Create and clean up schema
    schema = _create_schema_for_function(task_function)
    schema = _remove_attributes_from_descriptions(schema)
    schema = _remove_top_level_single_element_allof(schema)

    # Include titles for custom-model-typed arguments
    schema = _include_titles(schema, verbose=verbose)

    # Include main title
    if schema.get("title") is None:

        def to_camel_case(snake_str):
            return "".join(x.capitalize() for x in snake_str.lower().split("_"))

        schema["title"] = to_camel_case(task_function.__name__)

    # Include descriptions of function. Note: this function works both
    # for usages 1 or 2 (see docstring).
    function_args_descriptions = _get_function_args_descriptions(
        package_name=package,
        module_path=executable,
        function_name=function_name,
        verbose=verbose,
    )

    schema = _insert_function_args_descriptions(
        schema=schema,
        descriptions=function_args_descriptions,
        verbose=verbose,
    )

    logging.info("[create_schema_for_single_task] END")
    return schema