| 105 | |
| 106 | |
| 107 | class PullRequestWorkflowBot: |
| 108 | |
| 109 | def __init__(self, event_name, event_payload, token=None, committers=None): |
| 110 | kwargs = {} |
| 111 | if token is not None: |
| 112 | kwargs["auth"] = github.Auth.Token(token) |
| 113 | self.github = github.Github(**kwargs) |
| 114 | self.event_name = event_name |
| 115 | self.event_payload = event_payload |
| 116 | self.committers = committers |
| 117 | |
| 118 | @cached_property |
| 119 | def pull(self): |
| 120 | """ |
| 121 | Returns a github.PullRequest object associated with the event. |
| 122 | """ |
| 123 | return self.repo.get_pull(self.event_payload['pull_request']['number']) |
| 124 | |
| 125 | @cached_property |
| 126 | def repo(self): |
| 127 | return self.github.get_repo(self.event_payload['repository']['id'], lazy=True) |
| 128 | |
| 129 | def is_committer(self, action): |
| 130 | """ |
| 131 | Returns whether the author of the action is a committer or not. |
| 132 | If the list of committer usernames is not available it will use the |
| 133 | author_association as a fallback mechanism. |
| 134 | """ |
| 135 | if self.committers: |
| 136 | return (self.event_payload[action]['user']['login'] in |
| 137 | self.committers) |
| 138 | return (self.event_payload[action]['author_association'] in |
| 139 | COMMITTER_ROLES) |
| 140 | |
| 141 | def handle(self): |
| 142 | current_state = None |
| 143 | try: |
| 144 | current_state = self.get_current_state() |
| 145 | except EventError: |
| 146 | # In case of error (more than one state) we clear state labels |
| 147 | # only possible if a label has been manually added. |
| 148 | self.clear_current_state() |
| 149 | next_state = self.compute_next_state(current_state) |
| 150 | if not current_state or current_state != next_state: |
| 151 | if current_state: |
| 152 | self.clear_current_state() |
| 153 | self.set_state(next_state) |
| 154 | |
| 155 | def get_current_state(self): |
| 156 | """ |
| 157 | Returns a PullRequestState with the current PR state label |
| 158 | based on label starting with LABEL_PREFIX. |
| 159 | If more than one label is found raises EventError. |
| 160 | If no label is found returns None. |
| 161 | """ |
| 162 | states = [label.name for label in self.pull.get_labels() |
| 163 | if label.name.startswith(LABEL_PREFIX)] |
| 164 | if len(states) > 1: |
no outgoing calls