prefect.utilities.validation

validate_schema

Validate that the provided schema is a valid json schema.

Parameters:

Name	Type	Description	Default
schema	dict	The schema to validate.	required

Raises:

Type	Description
ValueError	If the provided schema is not a valid json schema.

Source code in prefect/utilities/validation.py

def validate_schema(schema: dict):
    """
    Validate that the provided schema is a valid json schema.

    Args:
        schema: The schema to validate.

    Raises:
        ValueError: If the provided schema is not a valid json schema.

    """
    try:
        if schema is not None:
            # Most closely matches the schemas generated by pydantic
            jsonschema.Draft4Validator.check_schema(schema)
    except jsonschema.SchemaError as exc:
        raise ValueError(
            "The provided schema is not a valid json schema. Schema error:"
            f" {exc.message}"
        ) from exc

validate_values_conform_to_schema

Validate that the provided values conform to the provided json schema.

Parameters:

Name	Type	Description	Default
values	dict	The values to validate.	required
schema	dict	The schema to validate against.	required
ignore_required	bool	Whether to ignore the required fields in the schema. Should be used when a partial set of values is acceptable.	False

Raises:

Type	Description
ValueError	If the parameters do not conform to the schema.

Source code in prefect/utilities/validation.py

def validate_values_conform_to_schema(
    values: dict, schema: dict, ignore_required: bool = False
):
    """
    Validate that the provided values conform to the provided json schema.

    Args:
        values: The values to validate.
        schema: The schema to validate against.
        ignore_required: Whether to ignore the required fields in the schema. Should be
            used when a partial set of values is acceptable.

    Raises:
        ValueError: If the parameters do not conform to the schema.

    """
    if ignore_required:
        schema = remove_nested_keys(["required"], schema)

    try:
        if schema is not None and values is not None:
            jsonschema.validate(values, schema)
    except jsonschema.ValidationError as exc:
        if exc.json_path == "$":
            error_message = "Validation failed."
        else:
            error_message = (
                f"Validation failed for field {exc.json_path.replace('$.', '')!r}."
            )
        error_message += f" Failure reason: {exc.message}"
        raise ValueError(error_message) from exc
    except jsonschema.SchemaError as exc:
        raise ValueError(
            "The provided schema is not a valid json schema. Schema error:"
            f" {exc.message}"
        ) from exc