feat: Add filter-by-user functionality

current-implemented-CLI.md

@@ -6,11 +6,11 @@
 - `temp-dir` - The temporary directory to write JSON response files
 
 - `include-team` - Include a team to the query
+- `include-user` - Include a user to the query
 
 
 ## To Be Implemented
 
-- `include-user` - Include a specific user to the query
 - `include-repository` - Include specific repo to the query
 - `include-organization` - Include specific org to a query
 - `include-organization-repository` - Include specific org/repo to the query

src/queryFilter.py

@@ -0,0 +1,167 @@
+"""
+query_filter.py
+
+This module defines the QueryFilter class, which is used to filter GitHub projects based on various criteria.
+It is used in the Version2Query class to filter projects based on user input.
+"""
+
+class QueryFilter:
+  def __init__(
+    self,
+    include_teams:list[str]=None,
+    include_users:list[str]=None,
+    include_repositories:list[str]=None,
+    include_organizations:list[str]=None,
+    include_organization_repositories:list[str]=None,
+    include_labels:list[str]=None,
+    exclude_teams:list[str]=None,
+    exclude_users:list[str]=None,
+    exclude_repositories:list[str]=None,
+    exclude_organizations:list[str]=None,
+    exclude_organization_repositories:list[str]=None,
+    exclude_label:list[str]=None
+  ):
+    self.include_teams:list[str]                     = include_teams
+    self.include_users:list[str]                     = include_users
+    self.include_repositories:list[str]              = include_repositories
+    self.include_organizations:list[str]             = include_organizations
+    self.include_organization_repositories:list[str] = include_organization_repositories
+    self.include_labels:list[str]                    = include_labels
+    self.exclude_teams:list[str]                     = exclude_teams
+    self.exclude_users:list[str]                     = exclude_users
+    self.exclude_repositories:list[str]              = exclude_repositories
+    self.exclude_organizations:list[str]             = exclude_organizations
+    self.exclude_organization_repositories:list[str] = exclude_organization_repositories
+    self.exclude_label:list[str]                     = exclude_label
+
+  @property
+  def include_teams(self) -> list[str]:
+    return self._include_teams
+  @include_teams.setter
+  def include_teams(self, value:list[str]) -> None:
+    self._include_teams=value
+
+  @property
+  def include_users(self) -> list[str]:
+    return self._include_users
+  @include_users.setter
+  def include_users(self, value:list[str]) -> None:
+    self._include_users=value
+
+  @property
+  def include_repositories(self) -> list[str]:
+    return self._include_repositories
+  @include_repositories.setter
+  def include_repositories(self, value:list[str]) -> None:
+    self._include_repositories=value
+
+  @property
+  def include_organizations(self) -> list[str]:
+    return self._include_organizations
+  @include_organizations.setter
+  def include_organizations(self, value:list[str]) -> None:
+    self._include_organizations=value
+
+  @property
+  def include_organization_repositories(self) -> list[str]:
+    return self._include_organization_repositories
+  @include_organization_repositories.setter
+  def include_organization_repositories(self, value:list[str]) -> None:
+    self._include_organization_repositories=value
+
+  @property
+  def include_labels(self) -> list[str]:
+    return self._include_labels
+  @include_labels.setter
+  def include_labels(self, value:list[str]) -> None:
+    self._include_labels=value
+
+  @property
+  def exclude_teams(self) -> list[str]:
+    return self._exclude_teams
+  @exclude_teams.setter
+  def exclude_teams(self, value:list[str]) -> None:
+    self._exclude_teams=value
+
+  @property
+  def exclude_users(self) -> list[str]:
+    return self._exclude_users
+  @exclude_users.setter
+  def exclude_users(self, value:list[str]) -> None:
+    self._exclude_users=value
+
+  @property
+  def exclude_repositories(self) -> list[str]:
+    return self._exclude_repositories
+  @exclude_repositories.setter
+  def exclude_repositories(self, value:list[str]) -> None:
+    self._exclude_repositories=value
+
+  @property
+  def exclude_organizations(self) -> list[str]:
+    return self._exclude_organizations
+  @exclude_organizations.setter
+  def exclude_organizations(self, value:list[str]) -> None:
+    self._exclude_organizations=value
+
+  @property
+  def exclude_organization_repositories(self) -> list[str]:
+    return self._exclude_organization_repositories
+  @exclude_organization_repositories.setter
+  def exclude_organization_repositories(self, value:list[str]) -> None:
+    self._exclude_organization_repositories=value
+
+  @property
+  def exclude_label(self) -> list[str]:
+    return self._exclude_label
+  @exclude_label.setter
+  def exclude_label(self, value:list[str]) -> None:
+    self._exclude_label=value
+
+  def print_filters(self) -> None:
+    """Print the filters for debugging purposes."""
+    print(f"[bold green]Filters:[/bold green]")
+    print(f"Include Teams: {self.include_teams}")
+    print(f"Include Users: {self.include_users}")
+    print(f"Include Repositories: {self.include_repositories}")
+    print(f"Include Organizations: {self.include_organizations}")
+    print(f"Include Organization Repositories: {self.include_organization_repositories}")
+    print(f"Include Labels: {self.include_labels}")
+    print(f"Exclude Teams: {self.exclude_teams}")
+    print(f"Exclude Users: {self.exclude_users}")
+    print(f"Exclude Repositories: {self.exclude_repositories}")
+    print(f"Exclude Organizations: {self.exclude_organizations}")
+    print(f"Exclude Organization Repositories: {self.exclude_organization_repositories}")
+    print(f"Exclude Label: {self.exclude_label}")
+
+  def get_filters(self) -> dict[str,list[str]]:
+    """Get the filters as a dictionary."""
+    return {
+      "include_teams":                     self.include_teams,
+      "include_users":                     self.include_users,
+      "include_repositories":              self.include_repositories,
+      "include_organizations":             self.include_organizations,
+      "include_organization_repositories": self.include_organization_repositories,
+      "include_labels":                    self.include_labels,
+      "exclude_teams":                     self.exclude_teams,
+      "exclude_users":                     self.exclude_users,
+      "exclude_repositories":              self.exclude_repositories,
+      "exclude_organizations":             self.exclude_organizations,
+      "exclude_organization_repositories": self.exclude_organization_repositories,
+      "exclude_label":                     self.exclude_label
+    }
+
+  def set_filters_from_dict(self, filters:dict[str,list[str]]):
+    """Set the filters from a dictionary."""
+    self.include_teams                     = filters.get("include_teams", [])
+    self.include_users                     = filters.get("include_users", [])
+    self.include_repositories              = filters.get("include_repositories", [])
+    self.include_organizations             = filters.get("include_organizations", [])
+    self.include_organization_repositories = filters.get("include_organization_repositories", [])
+    self.include_labels                    = filters.get("include_labels", [])
+    self.exclude_teams                     = filters.get("exclude_teams", [])
+    self.exclude_users                     = filters.get("exclude_users", [])
+    self.exclude_repositories              = filters.get("exclude_repositories", [])
+    self.exclude_organizations             = filters.get("exclude_organizations", [])
+    self.exclude_organization_repositories = filters.get("exclude_organization_repositories", [])
+    self.exclude_label                     = filters.get("exclude_label", [])

src/version2config.py

@@ -50,7 +50,6 @@ def init_parser(self):
             dest="include_user",
             action="append",
             type=str,
-            nargs="+",
             help="Include all issues and PRs for the provided user [Required Parameter]"
         )
 

src/version2query.py

@@ -7,19 +7,26 @@
 from pathlib import Path
 from ghapi.all import GhApi
 from rich import print
+from .queryFilter import QueryFilter
 
 class Version2Query:
-  def __init__(self, temp_dir:str = "tmp.dir", output_file:str="output.items.json"):
+  def __init__(self, temp_dir:str="tmp.dir", output_file:str="output.items.json"):
     self.temp_dir:Path = Path(temp_dir)
     self.output_file:str = output_file
     self.api:GhApi = GhApi()
     self._validate_token()
+    self.filters:QueryFilter = QueryFilter()
 
   def _validate_token(self) -> None:
     """Validate Github token exists in the environment."""
     if not os.getenv("GITHUB_TOKEN"):
       raise ValueError("GITHUB_TOKEN environment variable is not set.")
 
+  def set_filters(self, filters:dict) -> None:
+    """Set filters for the query."""
+    self.filters.set_filters_from_dict(filters)
+    self.filters.print_filters()
+
   def get_github_token(self) -> str:
     return os.getenv("GITHUB_TOKEN")
 
@@ -81,6 +88,31 @@ def filter_projects_by_team(self, project_list:list[dict], teams:list[str]) -> l
       print(f"[green]Found {len(matching_projects)} matching projects for team '{team}'[/green]")
     return filtered_projects
 
+  def filter_items_by_user(self) -> None:
+    """Filter items by user."""
+
+    filtered_items:list[dict] = []
+
+    # pull the data out of output.projects.json and write to that file again
+    with open(self.output_file) as f:
+      items = json.load(f)
+      for item in items:
+        for user in self.filters.include_users:
+          if user in item.get("assignees", []):
+            filtered_items.append(item)
+            break # avoid appending the item multiple times if it matches multiple users.
+
+    # move self.output_file to output.items.json.tmp
+    tmp_file = self.output_file + ".tmp"
+
+    # write the new output_file with the filter_items_by_user
+    with open(self.output_file, "w") as f:
+      json.dump(filtered_items, f, indent=2)
+
+    # remove the tmp_file
+    if Path(tmp_file).exists():
+      os.remove(tmp_file)
+
   def fetch_project_items(self, projects:list[dict]) -> bool:
     """Fetch all items on each project."""
 
@@ -138,7 +170,7 @@ def cleanup(self) -> bool:
     if self.temp_dir.exists():
       try:
         shutil.rmtree(self.temp_dir)
-        print(f"[bold red]Cleaned up temporary files in {self.temp_dir}[/bold red]")
+        print(f"[bold purple]Cleaned up temporary files in {self.temp_dir}[/bold purple]")
       except Exception as e:
         print(f"[red]Error cleaning up temporary files: {e}[/red]")
         rv = False
@@ -147,14 +179,11 @@ def cleanup(self) -> bool:
 
     return rv
 
-  def process(self, teams:list[str] = None) -> bool:
+  def process(self) -> bool:
     """Main processing method for the Version2Query class."""
-    if not teams:
-      print("[red]No team names provided. Exiting...[/red]")
-      return False
 
     if self.temp_dir.exists():
-      cleanup()
+      self.cleanup()
 
     orgs = self.get_github_orgs()
     if not orgs:
@@ -166,10 +195,15 @@ def process(self, teams:list[str] = None) -> bool:
       print("[red]No projects found. Exiting...[/red]")
       return False
 
-    filtered_projects = self.filter_projects_by_team(all_projects, teams)
-    if not filtered_projects:
-      print("[red]No projects found matching the team names. Exiting...[/red]")
-      return False
+    # filter by team names
+    filtered_projects = all_projects
+    if self.filters.include_teams is not None:
+      teams = self.filters.include_teams
+      filtered_projects_by_teams = self.filter_projects_by_team(all_projects, teams)
+      if not filtered_projects_by_teams:
+        print("[red]No projects found matching the team names. Exiting...[/red]")
+        return False
+      filtered_projects = filtered_projects_by_teams
 
     if not self.fetch_project_items(filtered_projects):
       print("[red]Failed to fetch project items. Exiting...[/red]")
@@ -179,19 +213,24 @@ def process(self, teams:list[str] = None) -> bool:
       print("[red]Failed to consolidate items. Exiting...[/red]")
       return False
 
-    return self.cleanup()
+    # filter items by user
+    if self.filters.include_users is not None:
+      self.filter_items_by_user()
 
+    return self.cleanup()
 
+# This main method is used for testing out the version2query class
 def main():
   """The primary method for the version2query.py script."""
   teams:list = input("Enter team name(s) to filter projects: ").split(",")
   test_temp_dir:Path = Path(f"tmp.dir")
   test_output_file:str = f"output.items.json"
 
   query = Version2Query(temp_dir=test_temp_dir, output_file=test_output_file)
-  if not query.process(teams):
+  query.filters.include_teams = teams
+  if not query.process():
     print("[red]Processing failed.[/red]")
-    return
+    raise RunTimeError("Processing failed.")
 
 if __name__ == "__main__":
   main()

version2.py

@@ -10,14 +10,32 @@ def main():
     config.display_config()
     temp_dir:Path = Path(config.temp_dir)
     output_file:str = config.output_file
-    teams:list[str] = config.include_team
+
+    # Get parameters for filters
+    filters:dict[str,list[str]] = {
+      "include_teams":                     config.include_team,
+      "include_users":                     config.include_user,
+      "include_repositories":              config.include_repository,
+      "include_organizations":             config.include_organization,
+      "include_organization_repositories": config.include_organization_repository,
+      "include_labels":                    config.include_label,
+      "exclude_teams":                     config.exclude_team,
+      "exclude_users":                     config.exclude_user,
+      "exclude_repositories":              config.exclude_repository,
+      "exclude_organizations":             config.exclude_organization,
+      "exclude_organization_repositories": config.exclude_organization_repository,
+      "exclude_label":                     config.exclude_label
+    }
 
     query:VersionTwoQuery = Version2Query(temp_dir=temp_dir, output_file=output_file)
     ss_gen = StaticSiteGenerator()
 
+    # Set filters
+    query.set_filters(filters=filters)
+
     # Generate output_file or die
     logging.info("Querying GitHub API...")
-    if not query.process(teams):
+    if not query.process():
       logging.error("Failed to process query.")
       return