From 79cf4e58e43ac5237e1b55f6a15893ba5cd2eddc Mon Sep 17 00:00:00 2001 From: Campbell Barton Date: Wed, 28 Feb 2024 11:02:51 +1100 Subject: [PATCH] Cleanup: type check scripts in tools/triage --- tools/check_source/check_mypy_config.py | 3 - tools/triage/gitea_utils.py | 85 ++++++++++++++++--------- tools/triage/issues_needing_info.py | 2 +- tools/triage/weekly_report.py | 71 +++++++++++++-------- 4 files changed, 102 insertions(+), 59 deletions(-) diff --git a/tools/check_source/check_mypy_config.py b/tools/check_source/check_mypy_config.py index 44c6c451885..556425d3fcf 100644 --- a/tools/check_source/check_mypy_config.py +++ b/tools/check_source/check_mypy_config.py @@ -52,9 +52,6 @@ PATHS_EXCLUDE = set( "tools/check_source/check_descriptions.py", "tools/check_source/check_header_duplicate.py", "tools/check_source/check_unused_defines.py", - "tools/triage/gitea_utils.py", # TODO (low priority). - "tools/triage/issues_needing_info.py", # TODO (low priority). - "tools/triage/weekly_report.py", # TODO (low priority). "tools/utils/blend2json.py", "tools/utils/blender_keyconfig_export_permutations.py", "tools/utils/blender_merge_format_changes.py", diff --git a/tools/triage/gitea_utils.py b/tools/triage/gitea_utils.py index 91f7dd099aa..157c2fa4f77 100644 --- a/tools/triage/gitea_utils.py +++ b/tools/triage/gitea_utils.py @@ -5,15 +5,25 @@ # Simple module for inspecting GITEA users, pulls and issues. +import datetime import json import urllib.error import urllib.parse import urllib.request +from typing import ( + Any, + Dict, + List, + Optional, + Set, + Union, +) + BASE_API_URL = "https://projects.blender.org/api/v1" -def url_json_get(url): +def url_json_get(url: str) -> Optional[Union[Dict[str, Any], List[Dict[str, Any]]]]: try: # Make the HTTP request and store the response in a 'response' object response = urllib.request.urlopen(url) @@ -22,13 +32,19 @@ def url_json_get(url): print("Error making HTTP request:", ex) return None - # Convert the response content to a JSON object containing the user information - return json.loads(response.read()) + # Convert the response content to a JSON object containing the user information. + result = json.loads(response.read()) + assert result is None or isinstance(result, (dict, list)) + return result -def url_json_get_all_pages(url, limit=50, verbose=False): +def url_json_get_all_pages( + url: str, + limit: int = 50, + verbose: bool = False, +) -> List[Dict[str, Any]]: assert limit <= 50, "50 is the maximum limit of items per page" - result = [] + result: List[Dict[str, Any]] = [] page = 1 while True: if verbose: @@ -42,7 +58,7 @@ def url_json_get_all_pages(url, limit=50, verbose=False): if not result_page: break - + assert isinstance(result_page, list) result.extend(result_page) if len(result_page) < limit: @@ -53,51 +69,59 @@ def url_json_get_all_pages(url, limit=50, verbose=False): return result -def gitea_user_get(username): +def gitea_user_get(username: str) -> Dict[str, Any]: """ Get the user data as JSON from the user name. https://docs.gitea.com/api/next/#tag/user/operation/userGet """ url = f"{BASE_API_URL}/users/{username}" - return url_json_get(url) + result = url_json_get(url) + assert isinstance(result, dict) + return result -def gitea_json_issue_get(issue_fullname): +def gitea_json_issue_get(issue_fullname: str) -> Dict[str, Any]: """ Get issue/pull JSON data. :param issue_fullname: string in the format "{owner}/{repo}/issues/{number}" """ url = f"{BASE_API_URL}/repos/{issue_fullname}" - return url_json_get(url) + result = url_json_get(url) + assert isinstance(result, dict) + return result -def gitea_json_activities_get(username, date): +def gitea_json_activities_get(username: str, date: str) -> List[Dict[str, Any]]: """ List a user's activity feeds. :param username: username of user. :param date: the date of the activities to be found. """ activity_url = f"{BASE_API_URL}/users/{username}/activities/feeds?only-performed-by=true&date={date}" - return url_json_get_all_pages(activity_url) + result = url_json_get_all_pages(activity_url) + assert isinstance(result, list) + return result def gitea_json_issues_search( - type=None, - since=None, - before=None, - state='all', - labels=None, - created=False, - reviewed=False, - access_token=None, - verbose=True): + type: Optional[str] = None, + since: Optional[str] = None, + before: Optional[str] = None, + state: str = 'all', + labels: Optional[str] = None, + created: bool = False, + reviewed: bool = False, + access_token: Optional[str] = None, + verbose: bool = True, +) -> List[Dict[str, Any]]: """ Search for issues across the repositories that the user has access to. :param type: filter by type (issues / pulls) if set. :param since: Only show notifications updated after the given time. This is a timestamp in RFC 3339 format. :param before: Only show notifications updated before the given time. This is a timestamp in RFC 3339 format. :param state: whether issue is open or closed. - :param labels: comma separated list of labels. Fetch only issues that have any of this labels. Non existent labels are discarded. + :param labels: comma separated list of labels. + Fetch only issues that have any of this labels. Non existent labels are discarded. :param created: filter (issues / pulls) created by you, default is false. :param reviewed: filter pulls reviewed by you, default is false. :param access_token: token generated by the GITEA API. @@ -131,19 +155,20 @@ def gitea_json_issues_search( def gitea_json_issue_events_filter( - issue_fullname, - date_start=None, - date_end=None, - username=None, - labels=None, - event_type=set()): + issue_fullname: str, + date_start: Optional[datetime.datetime] = None, + date_end: Optional[datetime.datetime] = None, + username: Optional[str] = None, + labels: Optional[Set[str]] = None, + event_type: Set[str] = set(), +) -> List[Dict[str, Any]]: """ Filter all comments and events on the issue list. :param issue_fullname: string in the format "{owner}/{repo}/issues/{number}" :param date_start: if provided, only comments updated since the specified time are returned. :param date_end: if provided, only comments updated before the provided time are returned. :param labels: list of labels. Fetch only events that have any of this labels. - :param event_type: list of types of events in {"close", "commit_ref"...}. + :param event_type: set of types of events in {"close", "commit_ref"...}. :return: List of comments or events. """ issue_events_url = f"{BASE_API_URL}/repos/{issue_fullname}/timeline" @@ -180,7 +205,7 @@ def gitea_json_issue_events_filter( # WORKAROUND: This function doesn't involve GITEA, and the obtained username may not match the username used in GITEA. # However, it provides an option to fetch the configured username from the local Git, # in case the user does not explicitly supply the username. -def git_username_detect(): +def git_username_detect() -> Optional[str]: import os import subprocess diff --git a/tools/triage/issues_needing_info.py b/tools/triage/issues_needing_info.py index 178a964451a..8e7dde4c4e5 100644 --- a/tools/triage/issues_needing_info.py +++ b/tools/triage/issues_needing_info.py @@ -18,7 +18,7 @@ import datetime from gitea_utils import gitea_json_issues_search, gitea_json_issue_events_filter, git_username_detect -def print_needing_info_urls(username, before): +def print_needing_info_urls(username: str, before: str) -> None: print(f"Needs information from user before {before}:") diff --git a/tools/triage/weekly_report.py b/tools/triage/weekly_report.py index 2d4d19a1b27..7a97c6961c4 100644 --- a/tools/triage/weekly_report.py +++ b/tools/triage/weekly_report.py @@ -20,10 +20,24 @@ import argparse import datetime import json import re -from gitea_utils import gitea_json_activities_get, gitea_json_issue_get, gitea_json_issue_events_filter, gitea_user_get, git_username_detect + +from gitea_utils import ( + gitea_json_activities_get, + gitea_json_issue_events_filter, + gitea_json_issue_get, + gitea_user_get, git_username_detect, +) + +from typing import ( + Any, + Dict, + List, + Set, + Iterable, +) -def argparse_create(): +def argparse_create() -> argparse.ArgumentParser: parser = argparse.ArgumentParser( description="Generate Weekly Report", epilog="This script is typically used to help write weekly reports") @@ -53,36 +67,37 @@ def argparse_create(): return parser -def report_personal_weekly_get(username, start, verbose=True): +def report_personal_weekly_get(username: str, start: datetime.datetime, verbose: bool = True) -> None: - data_cache = {} + data_cache: Dict[str, Dict[str, Any]] = {} - def gitea_json_issue_get_cached(issue_fullname): + def gitea_json_issue_get_cached(issue_fullname: str) -> Dict[str, Any]: if issue_fullname not in data_cache: - data_cache[issue_fullname] = gitea_json_issue_get(issue_fullname) + issue = gitea_json_issue_get(issue_fullname) + data_cache[issue_fullname] = issue return data_cache[issue_fullname] - pulls_closed = set() - pulls_commented = set() - pulls_created = set() + pulls_closed: Set[str] = set() + pulls_commented: Set[str] = set() + pulls_created: Set[str] = set() - issues_closed = set() - issues_commented = set() - issues_created = set() + issues_closed: Set[str] = set() + issues_commented: Set[str] = set() + issues_created: Set[str] = set() - pulls_reviewed = [] + pulls_reviewed: List[str] = [] - issues_confirmed = [] - issues_needing_user_info = [] - issues_needing_developer_info = [] - issues_fixed = [] - issues_duplicated = [] - issues_archived = [] + issues_confirmed: List[str] = [] + issues_needing_user_info: List[str] = [] + issues_needing_developer_info: List[str] = [] + issues_fixed: List[str] = [] + issues_duplicated: List[str] = [] + issues_archived: List[str] = [] - commits_main = [] + commits_main: List[str] = [] - user_data = gitea_user_get(username) + user_data: Dict[str, Any] = gitea_user_get(username) for i in range(7): date_curr = start + datetime.timedelta(days=i) @@ -112,10 +127,16 @@ def report_personal_weekly_get(username, start, verbose=True): fullname = activity["repo"]["full_name"] + "/pulls/" + activity["content"].split('|')[0] pulls_reviewed.append(fullname) elif op_type == "commit_repo": - if activity["ref_name"] == "refs/heads/main" and activity["content"] and activity["repo"]["name"] != ".profile": + if ( + activity["ref_name"] == "refs/heads/main" and + activity["content"] and + activity["repo"]["name"] != ".profile" + ): content_json = json.loads(activity["content"]) + assert isinstance(content_json, dict) repo_fullname = activity["repo"]["full_name"] - for commits in content_json["Commits"]: + content_json_commits: List[Dict[str, Any]] = content_json["Commits"] + for commits in content_json_commits: # Skip commits that were not made by this user. Using email doesn't seem to # be possible unfortunately. if commits["AuthorName"] != user_data["full_name"]: @@ -207,7 +228,7 @@ def report_personal_weekly_get(username, start, verbose=True): print() # Print review stats - def print_pulls(pulls): + def print_pulls(pulls: Iterable[str]) -> None: for pull in pulls: pull_data = gitea_json_issue_get_cached(pull) title = pull_data["title"] @@ -232,7 +253,7 @@ def report_personal_weekly_get(username, start, verbose=True): if verbose: # Debug - def print_links(issues): + def print_links(issues: Iterable[str]) -> None: for fullname in issues: print(f"https://projects.blender.org/{fullname}")