def run_push_existing_classification()

in mozci/data/sources/taskcluster/__init__.py [0:0]
34 lines of code
6 McCabe index (conditional complexity)

    def run_push_existing_classification(self, branch, rev, environment):
        # Non-production environments are exposed in sub routes
        route_prefix = (
            "project.mozci.classification"
            if environment == "production"
            else f"project.mozci.{environment}.classification"
        )

        # We use buildUrl and manual requests.get instead of directly findArtifactFromTask from the taskcluster library
        # because the taskcluster library fails with redirects (https://github.com/taskcluster/taskcluster/issues/4998).
        try:
            # Proxy authentication does not seem to work here
            index = Index({"rootUrl": taskcluster.COMMUNITY_TASKCLUSTER_ROOT_URL})
            url = index.buildUrl(
                "findArtifactFromTask",
                f"{route_prefix}.{branch}.revision.{rev}",
                "public/classification.json",
            )
        except TaskclusterRestFailure as e:
            raise ContractNotFilled(
                self.name,
                "push_existing_classification",
                f"Failed to load existing classification for {branch} {rev}: {e}",
            )

        try:
            r = requests.get(url, allow_redirects=True)
            r.raise_for_status()
        except requests.exceptions.HTTPError as e:
            raise ContractNotFilled(
                self.name,
                "push_existing_classification",
                f"Failed to load existing classification for {branch} {rev}: {e}",
            )

        try:
            return r.json()["push"]["classification"]
        except KeyError:
            raise ContractNotFilled(
                self.name, "push_existing_classification", "Invalid classification data"
            )