(self, pr_number: int)
return EventInfoResponse(
            config=cfg.parsed,
            config_str=cfg.text,
            config_file_expression=cfg.file_expression,
            pull_request=pr,
            repository=RepoInfo(
                merge_commit_allowed=repository.get("mergeCommitAllowed", False),
                rebase_merge_allowed=repository.get("rebaseMergeAllowed", False),
                squash_merge_allowed=repository.get("squashMergeAllowed", False),
                delete_branch_on_merge=repository.get("deleteBranchOnMerge") is True,
                is_private=repository.get("isPrivate") is True,
            ),
            subscription=subscription,
            branch_protection=branch_protection,
            review_requests=get_requested_reviews(pr=pull_request),
            bot_reviews=bot_reviews,
            status_contexts=get_status_contexts(pr=pull_request),
            commits=get_commits(pr=pull_request),
            check_runs=get_check_runs(pr=pull_request),
            head_exists=get_head_exists(pr=pull_request),
            valid_merge_methods=get_valid_merge_methods(repo=repository),
        )
Retrieve all the information we need to evaluate a pull request

        This is basically the "do-all-the-things" query
Retrieve all the information we need to evaluate a pull request
Retrieve all the information we need to evaluate a pull request
async def get_event_info(self, pr_number: int) -> Optional[EventInfoResponse]:
        """
        Retrieve all the information we need to evaluate a pull request

        This is basically the "do-all-the-things" query
        """

        log = self.log.bind(pr=pr_number)

        api_features = await self.get_api_features()

        res = await self.send_query(
            query=get_event_info_query(
                requires_conversation_resolution=api_features.requires_conversation_resolution
                if api_features
                else True
            ),
            variables=dict(owner=self.owner, repo=self.repo, PRNumber=pr_number),
            installation_id=self.installation_id,
        )
        if res is None:
            return None

        data = res.get("data")
        if data is None:
            log.error("could not fetch event info", res=res)
            return None

        repository = get_repo(data=data)
        if not repository:
            log.warning("could not find repository")
            return None

        org_repo_default_branch = get_org_config_default_branch(data=data)

        subscription = (
            await self.get_subscription() if conf.SUBSCRIPTIONS_ENABLED else None
        )

        pull_request = get_pull_request(repo=repository)
        if not pull_request:
            log.warning("Could not find PR")
            return None

        # update the dictionary to match what we need for parsing
        pull_request["labels"] = get_labels(pr=pull_request)
        pull_request["latest_sha"] = get_sha(pr=pull_request)
        pull_request["number"] = pr_number
        try:
            pr = PullRequest.parse_obj(pull_request)
        except ValueError:
            log.warning("Could not parse pull request")
            return None

        cfg = await self.get_config_for_ref(
            ref=pr.baseRefName, org_repo_default_branch=org_repo_default_branch
        )
        if cfg is None:
            log.info("no config found")
            return None
        branch_protection = get_branch_protection(
            repo=repository, ref_name=pr.baseRefName
        )

        all_reviews = get_reviews(pr=pull_request)
        bot_reviews = self.get_bot_reviews(reviews=all_reviews)
        return EventInfoResponse(
            config=cfg.parsed,
            config_str=cfg.text,
            config_file_expression=cfg.file_expression,
            pull_request=pr,
            repository=RepoInfo(
                merge_commit_allowed=repository.get("mergeCommitAllowed", False),
                rebase_merge_allowed=repository.get("rebaseMergeAllowed", False),
                squash_merge_allowed=repository.get("squashMergeAllowed", False),
                delete_branch_on_merge=repository.get("deleteBranchOnMerge") is True,
                is_private=repository.get("isPrivate") is True,
            ),
            subscription=subscription,
            branch_protection=branch_protection,
            review_requests=get_requested_reviews(pr=pull_request),
            bot_reviews=bot_reviews,
            status_contexts=get_status_contexts(pr=pull_request),
            commits=get_commits(pr=pull_request),
            check_runs=get_check_runs(pr=pull_request),
            head_exists=get_head_exists(pr=pull_request),
            valid_merge_methods=get_valid_merge_methods(repo=repository),
        )
async def get_event_info ( self , pr_number : int ) -> Optional [ EventInfoResponse ] : log = self . log . bind ( pr = pr_number ) api_features = await self . get_api_features ( ) res = await self . send_query ( query = get_event_info_query ( requires_conversation_resolution = api_features . requires_conversation_resolution if api_features else True ) , variables = dict ( owner = self . owner , repo = self . repo , PRNumber = pr_number ) , installation_id = self . installation_id , ) if res is None : return None data = res . get ( "data" ) if data is None : log . error ( "could not fetch event info" , res = res ) return None repository = get_repo ( data = data ) if not repository : log . warning ( "could not find repository" ) return None org_repo_default_branch = get_org_config_default_branch ( data = data ) subscription = ( await self . get_subscription ( ) if conf . SUBSCRIPTIONS_ENABLED else None ) pull_request = get_pull_request ( repo = repository ) if not pull_request : log . warning ( "Could not find PR" ) return None # update the dictionary to match what we need for parsing pull_request [ "labels" ] = get_labels ( pr = pull_request ) pull_request [ "latest_sha" ] = get_sha ( pr = pull_request ) pull_request [ "number" ] = pr_number try : pr = PullRequest . parse_obj ( pull_request ) except ValueError : log . warning ( "Could not parse pull request" ) return None cfg = await self . get_config_for_ref ( ref = pr . baseRefName , org_repo_default_branch = org_repo_default_branch ) if cfg is None : log . info ( "no config found" ) return None branch_protection = get_branch_protection ( repo = repository , ref_name = pr . baseRefName ) all_reviews = get_reviews ( pr = pull_request ) bot_reviews = self . get_bot_reviews ( reviews = all_reviews ) return EventInfoResponse ( config = cfg . parsed , config_str = cfg . text , config_file_expression = cfg . file_expression , pull_request = pr , repository = RepoInfo ( merge_commit_allowed = repository . get ( "mergeCommitAllowed" , False ) , rebase_merge_allowed = repository . get ( "rebaseMergeAllowed" , False ) , squash_merge_allowed = repository . get ( "squashMergeAllowed" , False ) , delete_branch_on_merge = repository . get ( "deleteBranchOnMerge" ) is True , is_private = repository . get ( "isPrivate" ) is True , ) , subscription = subscription , branch_protection = branch_protection , review_requests = get_requested_reviews ( pr = pull_request ) , bot_reviews = bot_reviews , status_contexts = get_status_contexts ( pr = pull_request ) , commits = get_commits ( pr = pull_request ) , check_runs = get_check_runs ( pr = pull_request ) , head_exists = get_head_exists ( pr = pull_request ) , valid_merge_methods = get_valid_merge_methods ( repo = repository ) , )