add team_permissions to repository

Author: WolfgangFischerEtas

otterdog/models/github_organization.py

@@ -672,6 +672,7 @@ async def _process_single_repo(
     repo_name: str,
     jsonnet_config: JsonnetConfig,
     teams: dict[str, Any],
+    repo_permissions: dict[str, list[dict[str, Any]]],
     app_installations: dict[str, str],
 ) -> tuple[str, Repository]:
     rest_api = gh_client.rest_api
@@ -682,6 +683,8 @@ async def _process_single_repo(
 
     github_repo_workflow_data = await rest_api.repo.get_workflow_settings(github_id, repo_name)
     repo.workflows = RepositoryWorkflowSettings.from_provider_data(github_id, github_repo_workflow_data)
+    repo_permission = repo_permissions.get(repo_name, [])
+    repo.set_team_permissions({entry["name"]: entry["permission"] for entry in repo_permission})
 
     if jsonnet_config.default_branch_protection_rule_config is not None:
         # get branch protection rules of the repo
@@ -759,6 +762,30 @@ async def _process_single_repo(
     return repo_name, repo
 
 
+def build_repo_permissions(teams: list[dict[str, Any]]) -> dict[str, list[dict[str, Any]]]:
+    """
+    Convert the output from the graphql call, which is team-centric, to a repository
+    centric structure.
+    """
+
+    repo_permissions: dict[str, list[dict[str, Any]]] = {}
+    for team in teams:
+        team_slug = team["slug"]
+
+        # List of repo edges of this team
+        edges = team.get("repositories", {}).get("edges", [])
+
+        for edge in edges:
+            repo_name = edge["node"]["name"]
+            permission = edge["permission"]
+
+            if repo_name not in repo_permissions:
+                repo_permissions[repo_name] = []
+
+            repo_permissions[repo_name].append({"name": team_slug, "permission": permission})
+    return repo_permissions
+
+
 async def _load_repos_from_provider(
     github_id: str,
     org_settings: OrganizationSettings,
@@ -776,6 +803,7 @@ async def _load_repos_from_provider(
         repo_names = fnmatch.filter(repo_names, repo_filter)
 
     teams = {str(team["id"]): f"{github_id}/{team['slug']}" for team in await provider.get_org_teams(github_id)}
+    repo_permissions = build_repo_permissions(await provider.get_team_permissions(github_id))
 
     # limit the number of repos that are processed concurrently to avoid hitting secondary rate limits
     sem = asyncio.Semaphore(50 if concurrency is None else concurrency)
@@ -789,6 +817,7 @@ async def safe_process(repo_name):
                 repo_name,
                 jsonnet_config,
                 teams,
+                repo_permissions,
                 app_installations,
             )
 

otterdog/models/repository.py

@@ -107,6 +107,7 @@ class Repository(ModelObject):
     fork_default_branch_only: bool = dataclasses.field(metadata={"model_only": True})
 
     workflows: RepositoryWorkflowSettings = dataclasses.field(metadata={"embedded_model": True})
+    team_permissions: dict[str, str] | None
 
     # model only fields
     aliases: list[str] = dataclasses.field(metadata={"model_only": True}, default_factory=list)
@@ -190,6 +191,19 @@ class Repository(ModelObject):
         "rust",
     }
 
+    _valid_team_permissions: ClassVar[dict[str, str]] = {
+        "pull": "pull",
+        "triage": "triage",
+        "push": "push",
+        "maintain": "maintain",
+        "admin": "admin",
+        "READ": "pull",
+        "TRIAGE": "triage",
+        "WRITE": "push",
+        "MAINTAIN": "maintain",
+        "ADMIN": "admin",
+    }
+
     @property
     def model_object_name(self) -> str:
         return "repository"
@@ -245,6 +259,11 @@ def add_environment(self, environment: Environment) -> None:
     def set_environments(self, environments: list[Environment]) -> None:
         self.environments = environments
 
+    def set_team_permissions(self, permissions: dict[str, str]) -> None:
+        self.team_permissions = {
+            team: self._valid_team_permissions.get(perm, perm) for team, perm in permissions.items()
+        }
+
     def coerce_from_org_settings(self, org_settings: OrganizationSettings, for_patch: bool = False) -> Repository:
         copy = dataclasses.replace(self)
 
@@ -454,6 +473,16 @@ def validate(self, context: ValidationContext, parent_object: Any) -> None:
                         f"{self.get_model_header()} defines an unknown custom property with key '{k}'.",
                     )
 
+        if is_set_and_present(self.team_permissions):
+            for k, v in self.team_permissions.items():
+                if v not in self._valid_team_permissions:
+                    context.add_failure(
+                        FailureType.ERROR,
+                        f"invalid permission '{v}' "
+                        f"for team '{k}', allowed values are "
+                        f"('read/pull' | 'triage' | 'write/push' | 'maintain' | 'admin').",
+                    )
+
         for webhook in self.webhooks:
             webhook.validate(context, self)
 
@@ -854,9 +883,13 @@ def property_list_to_map(properties):
 
             return output
 
+        def transform_perm(permissions):
+            return {team: cls._valid_team_permissions[perm] for team, perm in permissions}
+
         mapping.update(
             {
                 "custom_properties": OptionalS("custom_properties", default={}) >> F(property_list_to_map),
+                "team_permissions": OptionalS("team_permissions", default={}) >> F(transform_perm),
                 "webhooks": K([]),
                 "secrets": K([]),
                 "variables": K([]),
@@ -1318,6 +1351,102 @@ def _include_gh_pages_patch_required_properties(
 
         return patch
 
+    @classmethod
+    def _calculate_team_permissions(
+        cls, patch: LivePatch[Repository]
+    ) -> tuple[list[str], dict[str, str], dict[str, str]]:
+        """
+        Computes the differences between the current and desired team permissions
+        for a repository based on the provided patch object.
+
+        The patch contains two dictionaries for team_permissions:
+            - 'from_perms': the current team-permission mapping in the system
+            - 'to_perms':   the desired target mapping
+
+        By comparing these two states, the function determines three categories
+        of changes that must be applied:
+
+        Returns:
+            deletes (list[str]):
+                A list of team names that exist in the current state but not in
+                the target state. These teams must have their permissions removed.
+
+            updates (dict[str, str]):
+                A mapping of team names to new permissions for teams that exist
+                in both states but whose permission value has changed.
+
+            adds (dict[str, str]):
+                A mapping of team names to permissions for teams that appear only
+                in the target state and therefore must be newly assigned.
+        """
+
+        changes = patch.changes
+        # No changes exist, so there is no "from" state.
+        # All permissions in expected_object.team_permissions must be added.
+        if changes is None:
+            expected = getattr(patch.expected_object, "team_permissions", None)
+            if isinstance(expected, dict):
+                return [], {}, dict(expected)  # only adds
+            return [], {}, {}
+
+        # Normal diff-based update
+        tp_change = changes.get("team_permissions")
+        if tp_change is None:
+            return [], {}, {}
+
+        from_value = tp_change.from_value
+        to_value = tp_change.to_value
+
+        if not isinstance(from_value, dict) or not isinstance(to_value, dict):
+            return [], {}, {}
+
+        # Now mypy knows these are dictionaries
+        from_perms: dict[str, str] = from_value
+        to_perms: dict[str, str] = to_value
+
+        deletes: list[str] = []
+        updates: dict[str, str] = {}
+        adds: dict[str, str] = {}
+
+        # Keys only in "from": delete
+        for team in from_perms.keys() - to_perms.keys():
+            deletes.append(team)
+
+        # Keys in both: update if permission changed
+        for team in from_perms.keys() & to_perms.keys():
+            if from_perms[team] != to_perms[team]:
+                updates[team] = to_perms[team]
+
+        # Keys only in "to": add
+        for team in to_perms.keys() - from_perms.keys():
+            adds[team] = to_perms[team]
+
+        return deletes, updates, adds
+
+    @classmethod
+    async def _apply_team_permission_changes(
+        cls,
+        provider,
+        org_id: str,
+        repo_name: str,
+        patch: LivePatch[Repository],
+    ) -> None:
+        """
+        Applies the calculated team-permission changes (delete, update, add) to the
+        given repository.
+        """
+
+        deletes, updates, adds = cls._calculate_team_permissions(patch)
+
+        for team in deletes:
+            await provider.delete_team_permission(org_id, repo_name, team)
+
+        for team, perm in updates.items():
+            await provider.update_team_permission(org_id, repo_name, team, perm)
+
+        for team, perm in adds.items():
+            await provider.add_team_permission(org_id, repo_name, team, perm)
+
     @classmethod
     async def apply_live_patch(
         cls,
@@ -1346,6 +1475,15 @@ async def apply_live_patch(
                         await expected_object.workflows.dict_to_provider_data(org_id, workflow_data, provider),
                     )
 
+                # Team permissions are defined on the repository but originate from the team side.
+                # A newly created repository always starts without any team permissions, even if
+                # the desired state already specifies them. After creation we therefore reconcile
+                # the permissions explicitly: remove teams that should not have access, update
+                # teams whose permission differs, and add teams that should be granted access.
+                # All three cases are kept to maintain a consistent reconciliation flow and to
+                # safely handle any unexpected initial state.
+                await cls._apply_team_permission_changes(provider, org_id, expected_object.name, patch)
+
             case LivePatchType.REMOVE:
                 await provider.delete_repo(org_id, unwrap(patch.current_object).name)
 
@@ -1369,3 +1507,11 @@ async def apply_live_patch(
                     github_data = await RepositoryWorkflowSettings.dict_to_provider_data(org_id, data, provider)
 
                     await provider.update_repo_workflow_settings(org_id, expected_object.name, github_data)
+
+                # Team permissions sit at the intersection of repositories and teams.
+                # A change in `team_permissions` can represent three different operations:
+                # - removing a team's permission (delete)
+                # - modifying an existing permission (update)
+                # - assigning a new permission (add)
+                # The patch is therefore decomposed into these three categories and applied here.
+                await cls._apply_team_permission_changes(provider, org_id, expected_object.name, patch)

otterdog/providers/github/__init__.py

@@ -422,6 +422,18 @@ async def add_repo_variable(self, org_id: str, repo_name: str, data: dict[str, s
     async def delete_repo_variable(self, org_id: str, repo_name: str, variable_name: str) -> None:
         await self.rest_api.repo.delete_variable(org_id, repo_name, variable_name)
 
+    async def get_team_permissions(self, org_id: str) -> list[dict[str, Any]]:
+        return await self.graphql_client.get_team_permissions(org_id)
+
+    async def update_team_permission(self, org_id: str, repo_name: str, team_name: str, team_permission: str) -> None:
+        await self.rest_api.repo.update_team_permission(org_id, repo_name, team_name, team_permission)
+
+    async def add_team_permission(self, org_id: str, repo_name: str, team_name: str, team_permission: str) -> None:
+        await self.rest_api.repo.add_team_permission(org_id, repo_name, team_name, team_permission)
+
+    async def delete_team_permission(self, org_id: str, repo_name: str, team_name: str) -> None:
+        await self.rest_api.repo.delete_team_permission(org_id, repo_name, team_name)
+
     async def dispatch_workflow(self, org_id: str, repo_name: str, workflow_name: str) -> bool:
         return await self.rest_api.repo.dispatch_workflow(org_id, repo_name, workflow_name)
 

otterdog/providers/github/graphql.py

@@ -255,11 +255,46 @@ async def get_team_membership(self, org_id: str, user_login: str) -> list[dict[s
         variables = {"owner": org_id, "user": user_login}
         return await self._run_paged_query(variables, "get-team-membership.gql", "data.organization.teams")
 
+    async def get_team_permissions(self, org_id: str) -> list[dict[str, Any]]:
+        _logger.debug(f"retrieving team permissions in org '{org_id}'")
+
+        variables = {"org": org_id}
+        # Run the graphql query with a limit of 100 for teams and repositories. If there are more than
+        # 100 teams available this gets handled in _run_paged_query. If there are more than 100
+        # permissions to repositories available then these are handled in the subsequent loop, where the
+        # pageInfo is used to get the missing repository entries.
+        teams = await self._run_paged_query(
+            input_variables=variables,
+            query_file="get-team-permissions-repositories.gql",
+            prefix_selector="data.organization.teams",
+        )
+        for team in teams:
+            repos = team["repositories"]["edges"]
+            page_info = team["repositories"]["pageInfo"]
+            if not page_info["hasNextPage"]:
+                continue
+            repo_cursor = page_info["endCursor"]
+            sub_vars = {
+                "org": org_id,
+                "teamSlug": team["slug"],
+                "endCursor": repo_cursor,
+            }
+            sub_result = await self._run_paged_query(
+                input_variables=sub_vars,
+                query_file="get-repository-permissions-of-team.gql",
+                prefix_selector="data.organization.team.repositories",
+                selector_type=".edges",
+            )
+            repos.extend(sub_result)
+
+        return teams
+
     async def _run_paged_query(
         self,
         input_variables: dict[str, Any],
         query_file: str,
         prefix_selector: str = "data.repository.branchProtectionRules",
+        selector_type: str = ".nodes",
     ) -> list[dict[str, Any]]:
         _logger.debug(f"running graphql query '{query_file}' with input '{json.dumps(input_variables)}'")
 
@@ -280,7 +315,7 @@ async def _run_paged_query(
                 _logger.trace("graphql result = %s", json.dumps(json_data, indent=2))
 
             if status < 400 and "data" in json_data:
-                rules_result = query_json(prefix_selector + ".nodes", json_data)
+                rules_result = query_json(prefix_selector + selector_type, json_data)
 
                 for rule in rules_result:
                     result.append(rule)

otterdog/providers/github/rest/repo_client.py

@@ -741,6 +741,43 @@ async def delete_environment(self, org_id: str, repo_name: str, env_name: str) -
 
         _logger.debug("removed repo environment '%s'", env_name)
 
+    async def get_team_permissions(self, org_id: str, repo_name: str) -> list[dict[str, Any]]:
+        _logger.debug("retrieving teams with permissions for repo '%s/%s'", org_id, repo_name)
+
+        try:
+            return await self.requester.request_json("GET", f"/repos/{org_id}/{repo_name}/teams")
+        except GitHubException as ex:
+            raise RuntimeError(f"failed getting team permissions for repo '{org_id}/{repo_name}':\n{ex}") from ex
+
+    async def update_team_permission(self, org_id: str, repo_name: str, team_name: str, team_permission: str) -> None:
+
+        status, _ = await self.requester.request_raw(
+            "PUT",
+            f"/orgs/{org_id}/teams/{team_name}/repos/{org_id}/{repo_name}",
+            data=json.dumps({"permission": team_permission}),
+        )
+
+        if status != 204:
+            raise RuntimeError(f"failed to update team permission for team {team_name} on repo {repo_name}")
+
+        _logger.debug(f"updated team permission for team {team_name} on repo {repo_name}")
+
+    async def add_team_permission(self, org_id: str, repo_name: str, team_name: str, team_permission: str) -> None:
+        _logger.debug(f"adding team permission for team {team_name} on repo {repo_name}")
+        await self.update_team_permission(org_id, repo_name, team_name, team_permission)
+        _logger.debug("added team permisson for team {team_name} on repo {repo_name}")
+
+    async def delete_team_permission(self, org_id: str, repo_name: str, team_name: str) -> None:
+        _logger.debug(f"deleting team permission for team {team_name} on repo {repo_name}")
+        status, _ = await self.requester.request_raw(
+            "DELETE", f"/orgs/{org_id}/teams/{team_name}/repos/{org_id}/{repo_name}"
+        )
+
+        if status != 204:
+            raise RuntimeError(f"failed to delete team permission for team {team_name} on repo {repo_name}")
+
+        _logger.debug(f"removed team permission for team {team_name} on repo {repo_name}")
+
     async def _get_deployment_branch_policies(self, org_id: str, repo_name: str, env_name: str) -> list[dict[str, Any]]:
         _logger.debug("retrieving deployment branch policies for env '%s'", env_name)