feat: fetch review requests (#16)

.github/workflows/pylint.yml

@@ -1,7 +1,6 @@
 name: Pylint
 
 on: 
-  push:
   pull_request:
     branches: ['main']
 
@@ -20,7 +19,7 @@ jobs:
     - name: Install dependencies
       run: |
         python -m pip install --upgrade pip
-        pip install pylint
+        pip install pylint aiohttp urwid pyyaml
     - name: Analysing the code with pylint
       run: |
         pylint $(git ls-files '*.py')

.pylintrc

@@ -1,12 +1,7 @@
 [MAIN]
-init-hook='import sys; sys.path.append(".")'
-
-disable=
-    C0114, # missing docstring
-    C0115, 
-    C0116, 
-    W0718, 
-    W0719
+init-hook="import sys; sys.path.append('./src/pream_team')"
+
+disable= C0114, C0115, C0116, W0718, W0719, R0913, R0902, R0903
 
 # Analyse import fallback blocks. This can be used to support both Python 2 and
 # 3 compatible code, which means that the block might have code that exists

README.md

@@ -13,13 +13,17 @@ python3 -m pip install  pream-team --upgrade
 You need a GitHub personal access token with full repo scope
 and with admin org read access if you want to specify `org` value.
 
-Besides the token you also need to provide a list of github usernames. You can do that through command line:
+Besides the token you also need to provide a list of github usernames. You can do that through command line or yaml config file (see below).
+
+
+pream-team will also dispay a list of pull requests where you (username specified by 'me' value in cli or yaml) or your team (team name specified by 'my_team') has been added as a reviewer. You can see those by clicking on 'Review requested' button on the top right of the TUI.
 
 If you provide `org` value, pream-team will fetch only the prs for repos that belong to the org.
-If you provide `ne` value, pream team will print out your approval status for the
+If you provide `me` value, pream team will print out your approval status for the
 pull request ('v', '@', 'x' for approved, commented, chages requeted) followed by the number of approvals for the PR eg. `[v|2] [draft|repo-name] - pr title`. If you dont provide the `me` value, you will only get the number of approvals eg. `[2] [draft|repo-name] - pr title`
 
 
+Command line options:
 ```
 options:
   -h, --help            show this help message and exit
@@ -29,6 +33,8 @@ options:
   --token TOKEN         GitHub API token.
   --org ORG             GitHub organization name.
   --me ME               Your GitHub username.
+  --my_team MY_TEAM     name of your gh team. used to check for review requests that requested team review but
+                        not you explicitly.
   --file FILE           Path to YAML file containing 'names', 'days', 'token' and 'org' fields. (Note that command line
                         arguments override YAML file configuration)
 ```
@@ -40,6 +46,7 @@ org: some-org # optional
 token: some-token # required
 days-back: 25 # optional
 me: username # optional
+my_team: team-name # optional
 names: # required (at least one)
   - "Teamamte-username-1"
   - "Teamamte-username-2"

src/pream_team/__init__.py

@@ -1,7 +1,8 @@
-from typing import List, Tuple, Optional
+from typing import List, Optional
 
 import argparse
 import os
+import sys
 import yaml
 from pream_team.github_pr_fetcher import GitHubPRFetcher
 
@@ -12,11 +13,32 @@
 def initialize_cache_manager(cache_file_path: str) -> Optional[CacheManager]:
     if os.path.isdir(os.path.dirname(cache_file_path)):
         return CacheManager(cache_file_path)
-    else:
-        return None
-
-
-def parse_args() -> Tuple[str, Optional[str], List[str], int, str, bool, Optional[str]]:
+    return None
+
+
+class Config:
+    def __init__(
+        self,
+        token: str,
+        org_name: Optional[str],
+        usernames: List[str],
+        days_back: int,
+        cache_file_path: str,
+        update_on_startup: bool,
+        me: Optional[str],
+        my_team: Optional[str],
+    ):
+        self.token = token
+        self.org_name = org_name
+        self.usernames = usernames
+        self.days_back = days_back
+        self.cache_file_path = cache_file_path
+        self.update_on_startup = update_on_startup
+        self.me = me
+        self.my_team = my_team
+
+
+def parse_args() -> Config:
     parser = argparse.ArgumentParser(
         description="Fetch GitHub PRs for specific users from the past N days."
     )
@@ -27,70 +49,78 @@ def parse_args() -> Tuple[str, Optional[str], List[str], int, str, bool, Optiona
     parser.add_argument("--token", type=str, help="GitHub API token.")
     parser.add_argument("--org", type=str, help="GitHub organization name.")
     parser.add_argument("--me", type=str, help="Your GH account name.")
-
+    parser.add_argument(
+        "--my_team",
+        type=str,
+        help="name of your gh team. used to check for review requests that requested "
+        "team review but not you explicitly.",
+    )
     parser.add_argument(
         "--file",
         type=str,
         default=os.path.expanduser("~/.prs/config.yml"),
         help="Path to YAML file containing 'names', 'days', 'token' and 'org' fields. "
-        + "(Note that command line arguments override YAML file configuration)",
+        "(Note that command line arguments override YAML file configuration)",
     )
 
     args = parser.parse_args()
 
-    token, org_name, usernames, days_back, cache_file_path, update_on_startup, me = (
-        "",
-        None,
-        [],
-        30,
-        "",
-        True,
-        None,
+    config = Config(
+        token="",
+        org_name=None,
+        usernames=[],
+        days_back=30,
+        cache_file_path=os.path.join(os.environ["HOME"], ".prs/cache.json"),
+        update_on_startup=True,
+        me=None,
+        my_team=None,
     )
 
     if os.path.exists(args.file):
-        with open(args.file, "r") as f:
+        with open(args.file, "r", encoding="utf-8") as f:
             data = yaml.safe_load(f)
-            token = data.get("token", "")
-            org_name = data.get("org", None)
-            usernames = data.get("names", [])
-            days_back = data.get("days-back", 30)
-            me = data.get("me", None)
-            cache_file_path = data.get(
-                "cache_dir", os.environ["HOME"] + "/.prs/cache.json"
-            )
-            update_on_startup = data.get("update_on_startup", True)
-
-    if args.token:
-        token = args.token
-    if args.org:
-        org_name = args.org
-    if args.names:
-        usernames = args.names
-    if args.days:
-        days_back = args.days
-    if args.me:
-        me = args.me
-
-    if not usernames or not token:
+            config.token = data.get("token", "")
+            config.org_name = data.get("org", None)
+            config.usernames = data.get("names", [])
+            config.days_back = data.get("days-back", 30)
+            config.me = data.get("me", None)
+            config.my_team = data.get("my-team", None)
+            config.cache_file_path = data.get("cache_dir", config.cache_file_path)
+            config.update_on_startup = data.get("update_on_startup", True)
+
+    config.token = args.token or config.token
+    config.org_name = args.org or config.org_name
+    config.usernames = args.names or config.usernames
+    config.days_back = args.days or config.days_back
+    config.me = args.me or config.me
+    config.my_team = args.my_team or config.my_team
+
+    if not config.usernames or not config.token:
         print("Token and Usernames must be provided.")
-        exit(1)
+        sys.exit(1)
 
-    return token, org_name, usernames, days_back, cache_file_path, update_on_startup, me
+    return config
 
 
 def app_main():
-    token, org_name, usernames, days_back, cache_file_path, update_on_startup, me = (
-        parse_args()
-    )
+    config = parse_args()
 
-    ui = PreamTeamUI(f"Team PRs in the last {days_back} days")
-    cache = initialize_cache_manager(cache_file_path)
+    ui = PreamTeamUI(f"Team PRs in the last {config.days_back} days")
+    cache = initialize_cache_manager(config.cache_file_path)
 
     def fetcher_factory():
-        return GitHubPRFetcher(token, org_name, days_back)
-
-    app = PreamTeamApp(fetcher_factory, cache, ui, usernames, update_on_startup, me)
+        return GitHubPRFetcher(config.token, config.org_name, config.days_back)
+
+    app = PreamTeamApp(
+        fetcher_factory,
+        cache,
+        ui,
+        config.usernames,
+        config.update_on_startup,
+        config.me,
+        config.my_team,
+        config.days_back,
+    )
     app.run()
 
 

src/pream_team/cache_manager.py

@@ -1,7 +1,17 @@
-from typing import Dict, List
+from datetime import datetime, timedelta
+from typing import Any, Dict, List, Optional
 import json
+import sys
 
 
+CACHE_TIMESTAMP_FORMAT = "%Y-%m-%d %H:%M:%S"
+
+
+class CachedPrs:
+    def __init__(self, prs: List[Any], timestamp: datetime):
+        self.prs = prs
+        self.timestamp = timestamp
+
 
 class CacheManager:
     def __init__(self, cache_file_path: str):
@@ -19,29 +29,63 @@ def _load_cache(self) -> Dict[str, Dict]:
         Returns an empty dictionary if the file does not exist or is empty.
         """
         try:
-            with open(self.cache_file_path, "r") as file:
+            with open(self.cache_file_path, "r", encoding="utf-8") as file:
                 return json.load(file)
         except (FileNotFoundError, json.JSONDecodeError):
             return {}
 
-    def save_prs(self, user: str, prs: List[Dict], timestamp):
+    def save_prs(self, user: str, prs: List[Dict], timestamp: datetime):
         """
-        Save the PRs for a specific user to the cache file, including the timestamp of when the PRs were saved.
+        Save the PRs for a specific user to the cache file, including the
+        timestamp of when the PRs were saved.
         param: user: The username of the user for whom the PRs are being saved.
         param: prs: The PRs to be saved.
         param: timestamp: The timestamp of when the PRs were saved.
         """
-        self.cache[user] = {"timestamp": timestamp, "prs": prs}
+        self.cache[user] = {
+            "timestamp": timestamp.strftime(CACHE_TIMESTAMP_FORMAT),
+            "prs": prs,
+        }
         try:
-            with open(self.cache_file_path, "w") as file:
+            with open(self.cache_file_path, "w", encoding="utf-8") as file:
                 json.dump(self.cache, file, indent=4)
         except (FileNotFoundError, json.JSONDecodeError):
-            exit(1)
+            sys.exit(1)
 
-    def load_prs(self, user: str) -> Dict:
+    def load_prs(self, user: str) -> Optional[CachedPrs]:
         """
         Load the PRs for a specific user from the cache.
         Returns an empty dictionary if there is no cached data for the user.
         :param user: The username of the user for whom the PRs are being loaded.
         """
-        return self.cache.get(user, {})
+        data = self.cache.get(user, {})
+        timestamp = data.get("timestamp", "")
+
+        if not data:
+            return None
+
+        return CachedPrs(
+            data.get("prs", []),
+            timestamp=datetime.strptime(timestamp, CACHE_TIMESTAMP_FORMAT),
+        )
+
+    def clean_up(self, older_than: timedelta) -> None:
+        """
+        Remove any cached PRs that are older than the specified time
+        from the cache, and save the updated cache to the file.
+        :param older_than: The time period for which PRs should be
+        retained in the cache.
+        """
+        for user, data in self.cache.items():
+            timestamp = data["timestamp"]
+            if (
+                datetime.now() - datetime.strptime(timestamp, "%Y-%m-%d %H:%M:%S")
+                > older_than
+            ):
+                del self.cache[user]
+
+        try:
+            with open(self.cache_file_path, "w", encoding="utf-8") as file:
+                json.dump(self.cache, file, indent=4)
+        except (FileNotFoundError, json.JSONDecodeError):
+            sys.exit(1)