def _validate_jmespath_result()

in src/fmeval/data_loaders/json_parser.py [0:0]

• 49 lines of code
• 9 McCabe index (conditional complexity)

    def _validate_jmespath_result(result: Union[Any, List[Any], List[List[Any]]], args: ColumnParseArguments) -> None:
        """Validates that the JMESPath result is as expected.

        For dataset columns in COLUMNS_WITH_LISTS, if `args.dataset_mime_type` is MIME_TYPE_JSON, then `result` is
        expected to be a 2D array. If MIME_TYPE_JSON_LINES, then `result` is expected to be a 1D array (list).

        For all other dataset columns, if `args.dataset_mime_type` is MIME_TYPE_JSON, then `result` is expected to be
        a 1D array (list). If MIME_TYPE_JSON_LINES, then `result` is expected to be a single scalar value.

        :param result: JMESPath query result to be validated.
        :param args: See ColumnParseArguments docstring.
        """
        if args.dataset_mime_type == MIME_TYPE_JSON:
            require(
                result and isinstance(result, list),
                f"Expected to find a non-empty list of samples for the {args.column.value.name} column using "
                f"JMESPath '{args.jmespath_parser.expression}' on {args.dataset_name}.",
            )
            require(
                all(x is not None for x in result),  # explicitly using "is not None" since values like 0 are false-y
                f"Expected an array of non-null values using JMESPath '{args.jmespath_parser.expression}' for "
                f"the {args.column.value.name} column of dataset `{args.dataset_name}`, but found at least "
                "one value that is None.",
            )
            if args.column.value.name in COLUMNS_WITH_LISTS:
                require(
                    all(isinstance(x, list) for x in result),
                    f"Expected a 2D array using JMESPath '{args.jmespath_parser.expression}' on dataset "
                    f"`{args.dataset_name}` but found at least one non-list object.",
                )
            else:
                require(
                    all(not isinstance(x, list) for x in result),
                    f"Expected a 1D array using JMESPath '{args.jmespath_parser.expression}' on dataset "
                    f"`{args.dataset_name}`, where each element of the array is a sample's {args.column.value.name}, "
                    f"but found at least one nested array.",
                )
        elif args.dataset_mime_type == MIME_TYPE_JSONLINES:
            require(
                result is not None,
                f"Found no values using {args.column.value.name} JMESPath '{args.jmespath_parser.expression}' "
                f"on dataset `{args.dataset_name}`.",
            )
            if args.column.value.name in COLUMNS_WITH_LISTS:
                require(
                    isinstance(result, list),
                    f"Expected to find a List using JMESPath '{args.jmespath_parser.expression}' on a dataset line in "
                    f"`{args.dataset_name}`, but found a non-list object instead.",
                )
            else:
                require(
                    not isinstance(result, list),
                    f"Expected to find a single value using {args.column.value.name} JMESPath "
                    f"'{args.jmespath_parser.expression}' on a dataset line in "
                    f"dataset `{args.dataset_name}`, but found a list instead.",
                )
        else:  # pragma: no cover
            raise EvalAlgorithmInternalError(
                f"args.dataset_mime_type is {args.dataset_mime_type}, but only JSON " "and JSON Lines are supported."
            )