feat(api): Publish CODEOWNERS collection endpoint

src/sentry/api/serializers/models/projectcodeowners.py

@@ -1,13 +1,17 @@
 import logging
+from datetime import datetime
+from typing import TypedDict
 
 from sentry.api.serializers import Serializer, register, serialize
 from sentry.api.serializers.models.projectownership import (
     OwnershipRuleOwnerResponse,
     OwnershipRuleResponse,
     OwnershipSchemaResponse,
 )
+from sentry.api.validators.project_codeowners import CodeOwnersErrors, build_codeowners_associations
 from sentry.integrations.api.serializers.models.repository_project_path_config import (
     RepositoryProjectPathConfigSerializer,
+    RepositoryProjectPathConfigSerializerResponse,
 )
 from sentry.integrations.services.integration import integration_service
 from sentry.integrations.source_code_management.repository import RepositoryIntegration
@@ -43,8 +47,33 @@ def _serialize_ownership_schema(schema: OwnershipSchema) -> OwnershipSchemaRespo
     return {"$version": schema["$version"], "rules": serialized_rules}
 
 
+class EmptyOwnershipSchemaResponse(TypedDict):
+    """Legacy CODEOWNERS records can have an empty schema before it is built."""
+
+
+class ProjectCodeOwnersResponseOptional(TypedDict, total=False):
+    codeMapping: RepositoryProjectPathConfigSerializerResponse
+    ownershipSyntax: str
+    errors: CodeOwnersErrors
+    schema: OwnershipSchemaResponse | EmptyOwnershipSchemaResponse
+    codeOwnersUrl: str
+
+
+class ProjectCodeOwnersResponse(ProjectCodeOwnersResponseOptional):
+    id: str
+    raw: str
+    dateCreated: datetime
+    dateUpdated: datetime
+    dateSynced: datetime | None
+    codeMappingId: str
+    provider: str
+
+
+DEFAULT_CODEOWNERS_EXPAND = ("errors", "hasTargetingContext")
+
+
 @register(ProjectCodeOwners)
-class ProjectCodeOwnersSerializer(Serializer):
+class ProjectCodeOwnersSerializer(Serializer[ProjectCodeOwnersResponse]):
     def __init__(
         self,
         expand=None,
@@ -56,47 +85,43 @@ def get_attrs(self, item_list, user, **kwargs):
         integrations = {
             i.id: i
             for i in integration_service.get_integrations(
-                integration_ids=[i.repository_project_path_config.integration_id for i in item_list]
+                integration_ids=[
+                    item.repository_project_path_config.integration_id for item in item_list
+                ]
             )
         }
         for item in item_list:
             code_mapping = item.repository_project_path_config
             repository = code_mapping.project_repository.repository
 
-            integration = integrations[item.repository_project_path_config.integration_id]
-            install = integration.get_installation(
-                organization_id=item.repository_project_path_config.organization_id,
-            )
+            integration = integrations.get(code_mapping.integration_id)
+            provider = "unknown"
             codeowners_url = "unknown"
-            if item.repository_project_path_config.organization_integration_id and (
-                isinstance(install, RepositoryIntegration)
-            ):
+            if integration and code_mapping.organization_integration_id:
+                provider = integration.provider
                 try:
-                    codeowners_response = install.get_codeowner_file(
-                        repository, ref=code_mapping.default_branch
+                    install = integration.get_installation(
+                        organization_id=code_mapping.organization_id,
                     )
-                    if codeowners_response is not None:
-                        codeowners_url = codeowners_response["html_url"]
-
+                    if isinstance(install, RepositoryIntegration):
+                        codeowners_response = install.get_codeowner_file(
+                            repository, ref=code_mapping.default_branch
+                        )
+                        if codeowners_response is not None:
+                            codeowners_url = codeowners_response["html_url"]
                 except Exception:
                     logger.exception("Could not get CODEOWNERS URL. Continuing execution.")
 
             attrs[item] = {
-                "provider": (
-                    integration.provider
-                    if item.repository_project_path_config.organization_integration_id
-                    else "unknown"
-                ),
+                "provider": provider,
                 "codeMapping": code_mapping,
                 "codeOwnersUrl": codeowners_url,
             }
 
         return attrs
 
-    def serialize(self, obj, attrs, user, **kwargs):
-        from sentry.api.validators.project_codeowners import build_codeowners_associations
-
-        data = {
+    def serialize(self, obj, attrs, user, **kwargs) -> ProjectCodeOwnersResponse:
+        data: ProjectCodeOwnersResponse = {
             "id": str(obj.id),
             "raw": obj.raw,
             "dateCreated": obj.date_added,
@@ -107,13 +132,18 @@ def serialize(self, obj, attrs, user, **kwargs):
         }
 
         if "codeMapping" in self.expand:
-            config = attrs.get("codeMapping", {})
             data["codeMapping"] = serialize(
-                config, user=user, serializer=RepositoryProjectPathConfigSerializer()
+                attrs["codeMapping"],
+                user=user,
+                serializer=RepositoryProjectPathConfigSerializer(),
             )
 
         if "ownershipSyntax" in self.expand:
-            data["ownershipSyntax"] = convert_schema_to_rules_text(obj.schema)
+            data["ownershipSyntax"] = (
+                convert_schema_to_rules_text(obj.schema)
+                if obj.schema and "$version" in obj.schema
+                else ""
+            )
 
         if "errors" in self.expand:
             _, errors = build_codeowners_associations(obj.raw, obj.project)

src/sentry/api/validators/project_codeowners.py

@@ -1,8 +1,8 @@
 from __future__ import annotations
 
 from collections import defaultdict
-from collections.abc import Collection, Mapping, Sequence
-from typing import Any
+from collections.abc import Collection, Sequence
+from typing import TypedDict
 
 from django.db.models.functions import Lower
 
@@ -17,6 +17,14 @@
 from sentry.users.services.user.service import user_service
 
 
+class CodeOwnersErrors(TypedDict):
+    missing_user_emails: list[str]
+    missing_external_users: list[str]
+    missing_external_teams: list[str]
+    teams_without_access: list[str]
+    users_without_access: list[str]
+
+
 def find_missing_associations(
     parsed_items: Sequence[str],
     associated_items: Collection[str],
@@ -26,7 +34,7 @@ def find_missing_associations(
 
 def build_codeowners_associations(
     codeowners: str, project: Project
-) -> tuple[Mapping[str, Any], Mapping[str, Any]]:
+) -> tuple[dict[str, str], CodeOwnersErrors]:
     """
     Build a dict of {external_name: sentry_name} associations for a raw codeowners file.
     Returns only the actors that exist and have access to the project.
@@ -58,8 +66,8 @@ def build_codeowners_associations(
         external_actors = []
 
     # Convert CODEOWNERS into IssueOwner syntax
-    users_dict = {}
-    teams_dict = {}
+    users_dict: dict[str, str] = {}
+    teams_dict: dict[str, str] = {}
 
     teams_without_access = set()
     teams_without_access_external_names = set()
@@ -136,16 +144,16 @@ def build_codeowners_associations(
             teams_without_access.add(f"#{team.slug}")
             teams_without_access_external_names.update(team_ids_to_external_names[team.id])
 
-    emails_dict = {}
+    emails_dict: dict[str, str] = {}
     user_emails = set()
     for user in users:
         for user_email in user.emails:
             emails_dict[user_email] = user_email
             user_emails.add(user_email)
 
-    associations = {**users_dict, **teams_dict, **emails_dict}
+    associations: dict[str, str] = {**users_dict, **teams_dict, **emails_dict}
 
-    errors = {
+    errors: CodeOwnersErrors = {
         "missing_user_emails": find_missing_associations(emails, user_emails),
         "missing_external_users": find_missing_associations(
             usernames, set(associations.keys()) | users_without_access_external_names

src/sentry/apidocs/examples/codeowners_examples.py

@@ -0,0 +1,56 @@
+from drf_spectacular.utils import OpenApiExample
+
+PROJECT_CODEOWNERS_RESPONSE = {
+    "id": "42",
+    "raw": "src/* user@example.com",
+    "dateCreated": "2024-01-15T10:30:00Z",
+    "dateUpdated": "2024-01-15T10:30:00Z",
+    "dateSynced": "2024-01-15T10:30:00Z",
+    "codeMappingId": "7",
+    "provider": "github",
+    "errors": {
+        "missing_external_teams": [],
+        "missing_external_users": [],
+        "missing_user_emails": [],
+        "teams_without_access": [],
+        "users_without_access": [],
+    },
+    "schema": {
+        "$version": 1,
+        "rules": [
+            {
+                "matcher": {"type": "codeowners", "pattern": "src/*"},
+                "owners": [{"type": "user", "id": "1", "name": "user@example.com"}],
+            }
+        ],
+    },
+    "codeOwnersUrl": "https://github.com/example/repository/blob/main/CODEOWNERS",
+}
+
+PROJECT_CODEOWNERS_RESPONSE_WITH_OWNERSHIP_SYNTAX = {
+    **PROJECT_CODEOWNERS_RESPONSE,
+    "ownershipSyntax": "codeowners:src/* user@example.com\n",
+}
+
+LIST_PROJECT_CODEOWNERS = [
+    OpenApiExample(
+        "List a project's CODEOWNERS configurations",
+        value=[PROJECT_CODEOWNERS_RESPONSE],
+        status_codes=["200"],
+        response_only=True,
+    )
+]
+
+CREATE_PROJECT_CODEOWNERS = [
+    OpenApiExample(
+        "Create a CODEOWNERS configuration",
+        value={"raw": "src/* user@example.com", "codeMappingId": "7"},
+        request_only=True,
+    ),
+    OpenApiExample(
+        "Created CODEOWNERS configuration",
+        value=PROJECT_CODEOWNERS_RESPONSE_WITH_OWNERSHIP_SYNTAX,
+        status_codes=["201"],
+        response_only=True,
+    ),
+]

src/sentry/integrations/api/serializers/models/repository_project_path_config.py

@@ -1,14 +1,35 @@
+from typing import TypedDict
+
 from django.db.models import prefetch_related_objects
 
 from sentry.api.serializers import Serializer, register
-from sentry.integrations.api.serializers.models.integration import serialize_provider
+from sentry.integrations.api.serializers.models.integration import (
+    IntegrationProviderInfo,
+    serialize_provider,
+)
 from sentry.integrations.models.repository_project_path_config import RepositoryProjectPathConfig
 from sentry.integrations.services.integration import integration_service
 from sentry.integrations.services.integration.model import RpcIntegration
 
 
+class RepositoryProjectPathConfigSerializerResponse(TypedDict):
+    id: str
+    projectId: str
+    projectSlug: str
+    repoId: str
+    repoName: str
+    integrationId: str | None
+    provider: IntegrationProviderInfo | None
+    stackRoot: str
+    sourceRoot: str
+    defaultBranch: str | None
+    automaticallyGenerated: bool
+
+
 @register(RepositoryProjectPathConfig)
-class RepositoryProjectPathConfigSerializer(Serializer):
+class RepositoryProjectPathConfigSerializer(
+    Serializer[RepositoryProjectPathConfigSerializerResponse]
+):
     def get_attrs(self, item_list, user, **kwargs):
         if not item_list:
             return {}
@@ -51,7 +72,9 @@ def get_attrs(self, item_list, user, **kwargs):
             for item in item_list
         }
 
-    def serialize(self, obj, attrs, user, **kwargs):
+    def serialize(
+        self, obj, attrs, user, **kwargs
+    ) -> RepositoryProjectPathConfigSerializerResponse:
         integration = attrs.get("integration")
 
         provider = integration.get_provider() if integration else None
@@ -61,7 +84,7 @@ def serialize(self, obj, attrs, user, **kwargs):
         project = obj.project_repository.project
         repository = obj.project_repository.repository
 
-        return {
+        response: RepositoryProjectPathConfigSerializerResponse = {
             "id": str(obj.id),
             "projectId": str(project.id),
             "projectSlug": project.slug,
@@ -74,3 +97,4 @@ def serialize(self, obj, attrs, user, **kwargs):
             "defaultBranch": obj.default_branch,
             "automaticallyGenerated": obj.automatically_generated,
         }
+        return response