feat: log changes in users, teams and repos in separate dataclasses

Author: mxmehl

gh_org_mgr/_gh_org.py

@@ -24,6 +24,7 @@
 from jwt.exceptions import InvalidKeyError
 
 from ._gh_api import get_github_secrets_from_env, run_graphql_query
+from ._stats import OrgChanges
 
 
 @dataclass
@@ -48,6 +49,7 @@ class GHorg:  # pylint: disable=too-many-instance-attributes, too-many-lines
     configured_repos_collaborators: dict[str, dict[str, str]] = field(default_factory=dict)
     archived_repos: list[Repository] = field(default_factory=list)
     unconfigured_team_repo_permissions: dict[str, dict[str, str]] = field(default_factory=dict)
+    stats: OrgChanges = field(default_factory=OrgChanges)
 
     # Re-usable Constants
     TEAM_CONFIG_FIELDS: dict[str, dict[str, str | None]] = field(  # pylint: disable=invalid-name
@@ -315,6 +317,7 @@ def sync_org_owners(self, dry: bool = False, force: bool = False) -> None:
         for user in owners_add:
             if gh_user := self._resolve_gh_username(user, "<org owners>"):
                 logging.info("Adding user '%s' as organization owner", gh_user.login)
+                self.stats.add_owner(gh_user.login)
                 if not dry:
                     self.org.add_to_members(gh_user, "admin")
 
@@ -326,6 +329,7 @@ def sync_org_owners(self, dry: bool = False, force: bool = False) -> None:
                     "Will make them a normal member",
                     gh_user.login,
                 )
+                self.stats.degrade_owner(gh_user.login)
                 # Handle authenticated user being the same as the one you want to degrade
                 if self._is_user_authenticated_user(gh_user):
                     logging.warning(
@@ -374,6 +378,7 @@ def create_missing_teams(self, dry: bool = False):
                     parent_id = self.org.get_team_by_slug(self._sluggify_teamname(parent)).id
 
                     logging.info("Creating team '%s' with parent ID '%s'", team, parent_id)
+                    self.stats.create_team(team)
                     # NOTE: We do not specify any team settings (description etc)
                     # here, this will happen later
                     if not dry:
@@ -386,6 +391,7 @@ def create_missing_teams(self, dry: bool = False):
 
                 else:
                     logging.info("Creating team '%s' without parent", team)
+                    self.stats.create_team(team)
                     if not dry:
                         self.org.create_team(
                             team,
@@ -479,9 +485,10 @@ def sync_current_teams_settings(self, dry: bool = False) -> None:
                     team.name,
                 )
                 for setting, diff in differences.items():
-                    logging.info(
-                        "Setting '%s': '%s' --> '%s'", setting, diff["dict2"], diff["dict1"]
-                    )
+                    change_str = f"Setting '{setting}': '{diff['dict2']}' --> '{diff['dict1']}'"
+                    logging.info(change_str)
+                    self.stats.edit_team_config(team.name, new_config=change_str)
+
                 # Execute team setting changes
                 if not dry:
                     try:
@@ -612,6 +619,7 @@ def sync_teams_members(self, dry: bool = False) -> None:  # pylint: disable=too-
                             team.name,
                             config_role,
                         )
+                        self.stats.pending_team_member(team=team.name, user=gh_user.login)
                         continue
 
                     logging.info(
@@ -620,6 +628,7 @@ def sync_teams_members(self, dry: bool = False) -> None:  # pylint: disable=too-
                         team.name,
                         config_role,
                     )
+                    self.stats.add_team_member(team=team.name, user=gh_user.login)
                     if not dry:
                         self._add_or_update_user_in_team(team=team, user=gh_user, role=config_role)
 
@@ -634,6 +643,7 @@ def sync_teams_members(self, dry: bool = False) -> None:  # pylint: disable=too-
                         team.name,
                         config_role,
                     )
+                    self.stats.change_team_member_role(team=team.name, user=gh_user.login)
                     if not dry:
                         self._add_or_update_user_in_team(team=team, user=gh_user, role=config_role)
 
@@ -652,6 +662,7 @@ def sync_teams_members(self, dry: bool = False) -> None:  # pylint: disable=too-
                             gh_user.login,
                             team.name,
                         )
+                        self.stats.remove_team_member(team=team.name, user=gh_user.login)
                         if not dry:
                             team.remove_membership(gh_user)
                     else:
@@ -676,6 +687,7 @@ def get_unconfigured_teams(
             if delete_unconfigured_teams:
                 for team in unconfigured_teams:
                     logging.info("Deleting team '%s' as it is not configured locally", team.name)
+                    self.stats.delete_team(team=team.name, deleted=True)
                     if not dry:
                         team.delete()
             else:
@@ -685,6 +697,8 @@ def get_unconfigured_teams(
                     "configured locally: %s. Taking no action about these teams.",
                     ", ".join(unconfigured_teams_str),
                 )
+                for team in unconfigured_teams:
+                    self.stats.delete_team(team=team.name, deleted=False)
 
     def get_members_without_team(
         self, dry: bool = False, remove_members_without_team: bool = False
@@ -714,6 +728,7 @@ def get_members_without_team(
                         "Removing user '%s' from organisation as they are not member of any team",
                         user.login,
                     )
+                    self.stats.remove_member_without_team(user=user.login, removed=True)
                     if not dry:
                         self.org.remove_from_membership(user)
             else:
@@ -723,6 +738,8 @@ def get_members_without_team(
                     "member of any team: %s",
                     ", ".join(members_without_team_str),
                 )
+                for user in members_without_team:
+                    self.stats.remove_member_without_team(user=user.login, removed=False)
 
     # --------------------------------------------------------------------------
     # Repos
@@ -850,6 +867,7 @@ def sync_repo_permissions(self, dry: bool = False, ignore_archived: bool = False
                     team.name,
                     perm,
                 )
+                self.stats.change_repo_team_permissions(repo=repo.name, team=team.name, perm=perm)
                 if not dry:
                     # Update permissions or newly add a team to a repo
                     team.update_team_repository(repo, perm)
@@ -875,6 +893,10 @@ def sync_repo_permissions(self, dry: bool = False, ignore_archived: bool = False
                     self._document_unconfigured_team_repo_permissions(
                         team=team, team_permission=teams[team], repo_name=repo.name
                     )
+                    # Collect this status in the stats
+                    self.stats.document_unconfigured_team_permissions(
+                        team=team.name, repo=repo.name, perm=teams[team]
+                    )
                     # Abort handling the repo sync as we don't touch unconfigured teams
                     continue
                 # Handle: Team is configured, but contains no config
@@ -893,6 +915,7 @@ def sync_repo_permissions(self, dry: bool = False, ignore_archived: bool = False
                 # Remove if any mismatch has been found
                 if remove:
                     logging.info("Removing team '%s' from repository '%s'", team.name, repo.name)
+                    self.stats.remove_team_from_repo(repo=repo.name, team=team.name)
                     if not dry:
                         team.remove_from_repos(repo)
 
@@ -1338,5 +1361,6 @@ def sync_repo_collaborator_permissions(self, dry: bool = False):
                     )
 
                     # Remove collaborator
+                    self.stats.remove_repo_collaborator(repo=repo.name, user=username)
                     if not dry:
                         repo.remove_from_collaborators(username)

gh_org_mgr/_stats.py

@@ -0,0 +1,148 @@
+# SPDX-FileCopyrightText: 2025 DB Systel GmbH
+#
+# SPDX-License-Identifier: Apache-2.0
+
+"""Dataclasses and functions for statistics"""
+
+from dataclasses import dataclass, field
+
+
+@dataclass
+class TeamChanges:  # pylint: disable=too-many-instance-attributes
+    """Dataclass holding information about the changes made to a team"""
+
+    newly_created: bool = False
+    deleted: bool = False
+    unconfigured: bool = False
+    changed_config: list[str] = field(default_factory=list)
+    added_members: list[str] = field(default_factory=list)
+    changed_members_role: list[str] = field(default_factory=list)
+    removed_members: list[str] = field(default_factory=list)
+    pending_members: list[str] = field(default_factory=list)
+
+
+@dataclass
+class RepoChanges:  # pylint: disable=too-many-instance-attributes
+    """Dataclass holding information about the changes made to a repository"""
+
+    changed_permissions_for_teams: list[str] = field(default_factory=list)
+    removed_teams: list[str] = field(default_factory=list)
+    unconfigured_teams_with_permissions: list[str] = field(default_factory=list)
+    removed_collaborators: list[str] = field(default_factory=list)
+
+
+@dataclass
+class OrgChanges:  # pylint: disable=too-many-instance-attributes
+    """Dataclass holding general statistics about the changes made to the organization"""
+
+    added_owners: list[str] = field(default_factory=list)
+    degraded_owners: list[str] = field(default_factory=list)
+    teams: dict[str, TeamChanges] = field(default_factory=dict)
+    repos: dict[str, RepoChanges] = field(default_factory=dict)
+    members_without_team: list[str] = field(default_factory=list)
+    removed_members: list[str] = field(default_factory=list)
+
+    # --------------------------------------------------------------------------
+    # Owners
+    # --------------------------------------------------------------------------
+    def add_owner(self, user: str) -> None:
+        """User has been added as owner"""
+        self.added_owners.append(user)
+
+    def degrade_owner(self, user: str) -> None:
+        """User has been degraded from owner to member"""
+        self.degraded_owners.append(user)
+
+    # --------------------------------------------------------------------------
+    # Teams
+    # --------------------------------------------------------------------------
+    def update_team(self, team_name: str, **changes: bool | str | list[str]) -> None:
+        """Update team changes"""
+        # Initialise team if not present
+        if team_name not in self.teams:
+            self.teams[team_name] = TeamChanges()
+
+        implement_changes(dc_object=self.teams[team_name], **changes)
+
+    def create_team(self, team: str) -> None:
+        """Team has been created"""
+        self.update_team(team_name=team, newly_created=True)
+
+    def edit_team_config(self, team: str, new_config: str) -> None:
+        """Team config has been changed"""
+        self.update_team(team_name=team, changed_config=new_config)
+
+    def delete_team(self, team: str, deleted: bool) -> None:
+        """Teams are not configured"""
+        self.update_team(team_name=team, unconfigured=True, deleted=deleted)
+
+    # --------------------------------------------------------------------------
+    # Members
+    # --------------------------------------------------------------------------
+    def add_team_member(self, team: str, user: str) -> None:
+        """User has been added to team"""
+        self.update_team(team_name=team, added_members=user)
+
+    def change_team_member_role(self, team: str, user: str) -> None:
+        """User role has been changed in team"""
+        self.update_team(team_name=team, changed_members_role=user)
+
+    def pending_team_member(self, team: str, user: str) -> None:
+        """User has a pending invitation"""
+        self.update_team(team_name=team, pending_members=user)
+
+    def remove_team_member(self, team: str, user: str) -> None:
+        """User has been removed from team"""
+        self.update_team(team_name=team, removed_members=user)
+
+    def remove_member_without_team(self, user: str, removed: bool) -> None:
+        """User is not in any team"""
+        self.members_without_team.append(user)
+        if removed:
+            self.removed_members.append(user)
+
+    # --------------------------------------------------------------------------
+    # Repos
+    # --------------------------------------------------------------------------
+    def update_repo(self, repo_name: str, **changes: bool | str | list[str]) -> None:
+        """Update team changes"""
+        # Initialise repo if not present
+        if repo_name not in self.teams:
+            self.repos[repo_name] = RepoChanges()
+
+        implement_changes(dc_object=self.repos[repo_name], **changes)
+
+    def change_repo_team_permissions(self, repo: str, team: str, perm: str) -> None:
+        """Team permissions have been changed for a repo"""
+        self.update_repo(repo_name=repo, changed_permissions_for_teams=f"{team}: {perm}")
+
+    def remove_team_from_repo(self, repo: str, team: str) -> None:
+        """Team has been removed form a repo"""
+        self.update_repo(repo_name=repo, removed_teams=team)
+
+    def document_unconfigured_team_permissions(self, repo: str, team: str, perm: str) -> None:
+        """Unconfigured team has permissions on repo"""
+        self.update_repo(repo_name=repo, unconfigured_teams_with_permissions=f"{team}: {perm}")
+
+    def remove_repo_collaborator(self, repo: str, user: str) -> None:
+        """Remove collaborator"""
+        self.update_repo(repo_name=repo, removed_collaborators=user)
+
+
+def implement_changes(dc_object, **changes: bool | str | list[str]):
+    """Smartly add changes to a dataclass object"""
+    for attribute, value in changes.items():
+        current_value = getattr(dc_object, attribute)
+        # attribute is list
+        if isinstance(current_value, list):
+            # input change is list
+            if isinstance(value, list):
+                current_value.extend(value)
+            # input change is not list
+            else:
+                current_value.append(value)
+        # All other cases, bool
+        else:
+            setattr(dc_object, attribute, value)
+
+    return dc_object

gh_org_mgr/manage.py

@@ -149,6 +149,8 @@ def main():
         logging.debug("Final dataclass:\n%s", org.pretty_print_dataclass())
         org.ratelimit()
 
+        print(org.stats)
+
     # Setup Team command
     elif args.command == "setup-team":
         setup_team(team_name=args.name, config_path=args.config, file_path=args.file)