iterative/dvc

View on GitHub
dvc/repo/metrics/diff.py

Summary

Maintainability
A
0 mins
Test Coverage
from typing import TYPE_CHECKING, TypedDict, Union

from funcy import compact

from dvc.utils.diff import diff as _diff_dict
from dvc.utils.diff import format_dict

if TYPE_CHECKING:
    from dvc.repo import Repo

    from .show import Result


class DiffResult(TypedDict, total=False):
    errors: dict[str, Union[Exception, dict[str, Exception]]]
    diff: dict[str, dict[str, dict]]


def _diff(
    result: dict[str, "Result"],
    old_rev: str,
    new_rev: str,
    **kwargs,
) -> DiffResult:
    old = result.get(old_rev, {})
    new = result.get(new_rev, {})

    old_data = old.get("data", {})
    new_data = new.get("data", {})

    res = DiffResult()
    errors = res.setdefault("errors", {})

    if old_error := old.get("error"):
        errors[old_rev] = old_error
    else:
        errors[old_rev] = {f: d["error"] for f, d in old_data.items() if "error" in d}

    if new_error := new.get("error"):
        errors[new_rev] = new_error
    else:
        errors[new_rev] = {f: d["error"] for f, d in new_data.items() if "error" in d}

    diff_data = _diff_dict(format_dict(old_data), format_dict(new_data), **kwargs)
    res = DiffResult(errors=errors, diff=diff_data)
    res["errors"] = compact(res.get("errors", {}))  # type: ignore[assignment]
    return compact(res)  # type: ignore[no-any-return]


def diff(
    repo: "Repo",
    a_rev: str = "HEAD",
    b_rev: str = "workspace",
    all: bool = False,  # noqa: A002
    **kwargs,
) -> DiffResult:
    if repo.scm.no_commits:
        return {}

    metrics = repo.metrics.show(revs=[a_rev, b_rev], hide_workspace=False, **kwargs)
    return _diff(metrics, a_rev, b_rev, with_unchanged=all)