(queue: WebhookQueueProtocol, push_event: PushEvent)
Trigger evaluation of PRs that depend on the pushed branch.
Trigger evaluation of PRs that depend on the pushed branch.
Trigger evaluation of PRs that depend on the pushed branch .
async def push(queue: WebhookQueueProtocol, push_event: PushEvent) -> None:
    """
    Trigger evaluation of PRs that depend on the pushed branch.
    """
    owner = push_event.repository.owner.login
    repo = push_event.repository.name
    installation_id = str(push_event.installation.id)
    branch_name = get_branch_name(push_event.ref)
    log = logger.bind(ref=push_event.ref, branch_name=branch_name)
    if branch_name is None:
        log.info("could not extract branch name from ref")
        return
    async with Client(
        owner=owner, repo=repo, installation_id=installation_id
    ) as api_client:
        # find all the PRs that depend on the branch affected by this push and
        # queue them for evaluation.
        # Any PR that has a base ref matching our event ref is dependent.
        prs = await api_client.get_open_pull_requests(base=branch_name)
        if prs is None:
            log.info("api call to find pull requests failed")
            return None
        for pr in prs:
            await queue.enqueue(
                event=WebhookEvent(
                    repo_owner=owner,
                    repo_name=repo,
                    pull_request_number=pr.number,
                    target_name=pr.base.ref,
                    installation_id=installation_id,
                )
            )
async def push ( queue : WebhookQueueProtocol , push_event : PushEvent ) -> None : owner = push_event . repository . owner . login repo = push_event . repository . name installation_id = str ( push_event . installation . id ) branch_name = get_branch_name ( push_event . ref ) log = logger . bind ( ref = push_event . ref , branch_name = branch_name ) if branch_name is None : log . info ( "could not extract branch name from ref" ) return async with Client ( owner = owner , repo = repo , installation_id = installation_id ) as api_client : # find all the PRs that depend on the branch affected by this push and # queue them for evaluation. # Any PR that has a base ref matching our event ref is dependent. prs = await api_client . get_open_pull_requests ( base = branch_name ) if prs is None : log . info ( "api call to find pull requests failed" ) return None for pr in prs : await queue . enqueue ( event = WebhookEvent ( repo_owner = owner , repo_name = repo , pull_request_number = pr . number , target_name = pr . base . ref , installation_id = installation_id , ) )