def from_phabricator_trigger()

in bot/code_review_bot/revisions.py [0:0]

• 56 lines of code
• 10 McCabe index (conditional complexity)

    def from_phabricator_trigger(build_target_phid: str, phabricator: PhabricatorAPI):
        assert build_target_phid.startswith("PHID-HMBT-")

        # This is the very first call on Phabricator API for that build, so we need to retry
        # a few times as the revision may not be immediately public
        buildable = None
        for i in range(5):
            try:
                buildable = phabricator.find_target_buildable(build_target_phid)
                break
            except Exception as e:
                logger.info(
                    "Failed to load Harbormaster build on try {i+1}/5, will retry in 30 seconds",
                    error=str(e),
                )
                time.sleep(30)
        if buildable is None:
            raise Exception("Failed to load Habormaster build, no more tries left")

        diff_phid = buildable["fields"]["objectPHID"]
        assert diff_phid.startswith("PHID-DIFF-")

        # Load diff details to get the diff revision
        # We also load the commits list in order to get the email of the author of the
        # patch for sending email if builds are failing.
        diffs = phabricator.search_diffs(
            diff_phid=diff_phid, attachments={"commits": True}
        )
        assert len(diffs) == 1, f"No diff available for {diff_phid}"
        diff = diffs[0]
        logger.info("Found diff", id=diff["id"], phid=diff["phid"])
        revision_phid = diff["revisionPHID"]

        # Load revision details from Phabricator
        revision = phabricator.load_revision(revision_phid)
        logger.info("Found revision", id=revision["id"], phid=revision["phid"])

        # Lookup repository details and match with a known repo from configuration
        repo_phid = revision["fields"]["repositoryPHID"]
        repos = phabricator.request(
            "diffusion.repository.search", constraints={"phids": [repo_phid]}
        )
        assert (
            len(repos["data"]) == 1
        ), f"No repository found on Phabrictor for {repo_phid}"
        phab_repo = repos["data"][0]
        repo_name = phab_repo["fields"]["name"]
        known_repos = {r.name: r for r in settings.repositories}
        repository = known_repos.get(repo_name)
        if repository is None:
            raise Exception(
                f"No repository found in configuration for {repo_name} - {repo_phid}"
            )
        logger.info("Found repository", name=repo_name, phid=repo_phid)

        return Revision(
            phabricator_id=revision["id"],
            phabricator_phid=revision_phid,
            diff_id=diff["id"],
            diff_phid=diff["phid"],
            diff=diff,
            build_target_phid=build_target_phid,
            url="https://{}/D{}".format(phabricator.hostname, revision["id"]),
            revision=revision,
            base_changeset="tip",
            base_repository=repository.url,
            base_repository_conf=repository,
            repository_try_name=repository.try_name,
        )