def compute_is_valid_response()

in services/worker/src/worker/job_runners/config/is_valid.py [0:0]

• 45 lines of code
• 13 McCabe index (conditional complexity)

def compute_is_valid_response(dataset: str, config: str) -> tuple[IsValidResponse, float]:
    """
    Get the response of 'config-is-valid' for one specific dataset config on huggingface.co.

    A dataset config is valid if any of the artifacts for any of the
    steps is valid.
    Args:
        dataset (`str`):
            A namespace (user or an organization) and a repo name separated
            by a `/`.
        config (`str`):
            A configuration name.
    Raises:
        [~`libcommon.exceptions.PreviousStepFormatError`]:
          If the content of the previous step has not the expected format.

    Returns:
        `tuple[IsValidResponse, float]`: The response (viewer, preview, search, filter, statistics) and the progress.
    """
    logging.info(f"compute 'config-is-valid' response for {dataset=} {config=}")

    preview = False
    viewer = False
    search = False
    filter = False
    statistics = False

    try:
        split_names = get_split_names(dataset=dataset, config=config)
    except PreviousStepFormatError:
        # If the previous step did not return the expected content, we raise the error
        raise
    except Exception:
        logging.debug(
            "Erroneous response, or no response found, in previous step for this dataset: 'config-split-names'."
        )
        return IsValidResponse(
            preview=preview, viewer=viewer, search=search, filter=filter, statistics=statistics
        ), 0.0

    try:
        total = 0
        pending = 0
        for split in split_names:
            total += 1
            try:
                response = get_response(kind="split-is-valid", dataset=dataset, config=config, split=split)
            except CachedArtifactNotFoundError:
                logging.debug("No response found in previous step for this dataset: 'split-is-valid'.")
                pending += 1
                continue
            if response["http_status"] != HTTPStatus.OK:
                logging.debug(f"Previous step gave an error: {response['http_status']}.")
                continue
            split_is_valid_content = response["content"]
            preview = preview or split_is_valid_content["preview"]
            viewer = viewer or split_is_valid_content["viewer"]
            search = search or split_is_valid_content["search"]
            filter = filter or split_is_valid_content["filter"]
            statistics = statistics or split_is_valid_content["statistics"]
    except Exception as e:
        raise PreviousStepFormatError("Previous step did not return the expected content.", e) from e

    progress = (total - pending) / total if total else 1.0
    return (
        IsValidResponse(preview=preview, viewer=viewer, search=search, filter=filter, statistics=statistics),
        progress,
    )