|
30 | 30 | from scripts import install_third_party_libs |
31 | 31 |
|
32 | 32 | import requests |
33 | | -from typing import Any, Dict, List, Optional |
| 33 | +from typing import Dict, List, Optional, TypedDict |
| 34 | + |
| 35 | +# Define a TypedDict for pull request data. |
| 36 | +class PRData(TypedDict, total=False): |
| 37 | + number: int |
| 38 | + user: Dict[str, str] |
| 39 | + mergeable_state: str |
34 | 40 |
|
35 | 41 | # Global configuration. |
36 | 42 | GITHUB_TOKEN = os.getenv('GITHUB_TOKEN') |
@@ -65,39 +71,39 @@ def __init__(self, token: str, repo: str) -> None: |
65 | 71 | 'Accept': 'application/vnd.github.v3+json' |
66 | 72 | } |
67 | 73 |
|
68 | | - def list_open_prs(self) -> List[Dict[str, Any]]: |
| 74 | + def list_open_prs(self) -> List[PRData]: |
69 | 75 | """Fetches all open pull requests with pagination. |
70 | 76 |
|
71 | 77 | Returns: |
72 | | - List of dictionaries representing open pull requests. |
| 78 | + List of PRData dictionaries representing open pull requests. |
73 | 79 | """ |
74 | | - prs: List[Dict[str, Any]] = [] |
| 80 | + prs: List[PRData] = [] |
75 | 81 | page = 1 |
76 | 82 | while True: |
77 | 83 | url = f'{self.base_url}/pulls?state=open&page={page}&per_page=100' |
78 | 84 | response = requests.get(url, headers=self.rest_headers, timeout=TIMEOUT) |
79 | 85 | response.raise_for_status() |
80 | | - current_prs = response.json() |
| 86 | + current_prs: List[PRData] = response.json() |
81 | 87 | if not current_prs: |
82 | 88 | break |
83 | 89 | prs.extend(current_prs) |
84 | 90 | page += 1 |
85 | 91 | return prs |
86 | 92 |
|
87 | | - def fetch_pr_details(self, pr_number: int) -> Optional[Dict[str, Any]]: |
| 93 | + def fetch_pr_details(self, pr_number: int) -> Optional[PRData]: |
88 | 94 | """Fetches pull request details with retries until a definitive mergeable state is found. |
89 | 95 |
|
90 | 96 | Args: |
91 | 97 | pr_number: The number of the pull request. |
92 | 98 |
|
93 | 99 | Returns: |
94 | | - A dictionary with PR details if the mergeable state is determined; otherwise, None. |
| 100 | + A PRData dictionary with PR details if the mergeable state is determined; otherwise, None. |
95 | 101 | """ |
96 | 102 | pr_details_url = f'{self.base_url}/pulls/{pr_number}' |
97 | 103 | for attempt in range(RETRY_COUNT): |
98 | 104 | response = requests.get(pr_details_url, headers=self.rest_headers, timeout=TIMEOUT) |
99 | 105 | response.raise_for_status() |
100 | | - pr_details = response.json() |
| 106 | + pr_details: PRData = response.json() |
101 | 107 | mergeable_state = pr_details.get('mergeable_state') |
102 | 108 | if mergeable_state and mergeable_state != 'unknown': |
103 | 109 | return pr_details |
|
0 commit comments