ref(issues): Move project codeowners endpoints and tests to issues folder

src/sentry/api/urls.py

@@ -304,6 +304,8 @@
 from sentry.issues.endpoints.organization_issues_resolved_in_release import (
     OrganizationIssuesResolvedInReleaseEndpoint,
 )
+from sentry.issues.endpoints.project_codeowners_details import ProjectCodeOwnersDetailsEndpoint
+from sentry.issues.endpoints.project_codeowners_index import ProjectCodeOwnersEndpoint
 from sentry.issues.endpoints.project_user_issue import ProjectUserIssueEndpoint
 from sentry.issues.endpoints.team_all_unresolved_issues import TeamAllUnresolvedIssuesEndpoint
 from sentry.issues.endpoints.team_issue_breakdown import TeamIssueBreakdownEndpoint
@@ -548,7 +550,6 @@
 from .endpoints.catchall import CatchallEndpoint
 from .endpoints.check_am2_compatibility import CheckAM2CompatibilityEndpoint
 from .endpoints.chunk import ChunkUploadEndpoint
-from .endpoints.codeowners import ProjectCodeOwnersDetailsEndpoint, ProjectCodeOwnersEndpoint
 from .endpoints.custom_rules import CustomRulesEndpoint
 from .endpoints.data_scrubbing_selector_suggestions import DataScrubbingSelectorSuggestionsEndpoint
 from .endpoints.debug_files import (

src/sentry/issues/endpoints/project_codeowners_details.py

@@ -0,0 +1,113 @@
+from __future__ import annotations
+
+import logging
+from typing import Any
+
+from rest_framework import status
+from rest_framework.exceptions import PermissionDenied
+from rest_framework.request import Request
+from rest_framework.response import Response
+
+from sentry import analytics
+from sentry.analytics.events.codeowners_updated import CodeOwnersUpdated
+from sentry.api.api_owners import ApiOwner
+from sentry.api.api_publish_status import ApiPublishStatus
+from sentry.api.base import region_silo_endpoint
+from sentry.api.bases.project import ProjectEndpoint
+from sentry.api.exceptions import ResourceDoesNotExist
+from sentry.api.serializers import serialize
+from sentry.api.serializers.models import projectcodeowners as projectcodeowners_serializers
+from sentry.issues.serializers.codeowners import ProjectCodeOwnerSerializer, ProjectCodeOwnersMixin
+from sentry.models.project import Project
+from sentry.models.projectcodeowners import ProjectCodeOwners
+
+logger = logging.getLogger(__name__)
+
+
+@region_silo_endpoint
+class ProjectCodeOwnersDetailsEndpoint(ProjectEndpoint, ProjectCodeOwnersMixin):
+    owner = ApiOwner.ISSUES
+    publish_status = {
+        "DELETE": ApiPublishStatus.PRIVATE,
+        "PUT": ApiPublishStatus.PRIVATE,
+    }
+
+    def convert_args(
+        self,
+        request: Request,
+        organization_id_or_slug: int | str,
+        project_id_or_slug: int | str,
+        codeowners_id: str,
+        *args: Any,
+        **kwargs: Any,
+    ) -> tuple[Any, Any]:
+        args, kwargs = super().convert_args(
+            request, organization_id_or_slug, project_id_or_slug, *args, **kwargs
+        )
+        try:
+            kwargs["codeowners"] = ProjectCodeOwners.objects.get(
+                id=codeowners_id, project=kwargs["project"]
+            )
+        except ProjectCodeOwners.DoesNotExist:
+            raise ResourceDoesNotExist
+
+        return args, kwargs
+
+    def put(self, request: Request, project: Project, codeowners: ProjectCodeOwners) -> Response:
+        """
+        Update a CodeOwners
+        `````````````
+
+        :pparam string organization_id_or_slug: the id or slug of the organization.
+        :pparam string project_id_or_slug: the id or slug of the project to get.
+        :pparam string codeowners_id: id of codeowners object
+        :param string raw: the raw CODEOWNERS text
+        :param string codeMappingId: id of the RepositoryProjectPathConfig object
+        :auth: required
+        """
+        if not self.has_feature(request, project):
+            self.track_response_code("update", PermissionDenied.status_code)
+            raise PermissionDenied
+
+        serializer = ProjectCodeOwnerSerializer(
+            instance=codeowners,
+            context={"project": project},
+            partial=True,
+            data={**request.data},
+        )
+        if serializer.is_valid():
+            updated_codeowners = serializer.save()
+
+            user_id = getattr(request.user, "id", None) or None
+            analytics.record(
+                CodeOwnersUpdated(
+                    user_id=user_id,
+                    organization_id=project.organization_id,
+                    project_id=project.id,
+                    codeowners_id=updated_codeowners.id,
+                )
+            )
+            self.track_response_code("update", status.HTTP_200_OK)
+            return Response(
+                serialize(
+                    updated_codeowners,
+                    request.user,
+                    serializer=projectcodeowners_serializers.ProjectCodeOwnersSerializer(
+                        expand=["ownershipSyntax", "errors"]
+                    ),
+                ),
+                status=status.HTTP_200_OK,
+            )
+
+        self.track_response_code("update", status.HTTP_400_BAD_REQUEST)
+        return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST)
+
+    def delete(self, request: Request, project: Project, codeowners: ProjectCodeOwners) -> Response:
+        """
+        Delete a CodeOwners
+        """
+        if not self.has_feature(request, project):
+            raise PermissionDenied
+
+        codeowners.delete()
+        return Response(status=status.HTTP_204_NO_CONTENT)

src/sentry/issues/endpoints/project_codeowners_index.py

@@ -0,0 +1,137 @@
+import sentry_sdk
+from rest_framework import status
+from rest_framework.exceptions import PermissionDenied
+from rest_framework.request import Request
+from rest_framework.response import Response
+
+from sentry import analytics
+from sentry.analytics.events.codeowners_created import CodeOwnersCreated
+from sentry.api.api_owners import ApiOwner
+from sentry.api.api_publish_status import ApiPublishStatus
+from sentry.api.base import region_silo_endpoint
+from sentry.api.bases.project import ProjectEndpoint
+from sentry.api.serializers import serialize
+from sentry.api.serializers.models import projectcodeowners as projectcodeowners_serializers
+from sentry.api.validators.project_codeowners import validate_codeowners_associations
+from sentry.issues.ownership.grammar import (
+    convert_codeowners_syntax,
+    create_schema_from_issue_owners,
+)
+from sentry.issues.serializers.codeowners import ProjectCodeOwnerSerializer, ProjectCodeOwnersMixin
+from sentry.models.project import Project
+from sentry.models.projectcodeowners import ProjectCodeOwners
+
+
+@region_silo_endpoint
+class ProjectCodeOwnersEndpoint(ProjectEndpoint, ProjectCodeOwnersMixin):
+    owner = ApiOwner.ISSUES
+    publish_status = {
+        "GET": ApiPublishStatus.PRIVATE,
+        "POST": ApiPublishStatus.PRIVATE,
+    }
+
+    def refresh_codeowners_schema(self, codeowner: ProjectCodeOwners, project: Project) -> None:
+        if hasattr(codeowner, "schema") and (
+            codeowner.schema is None or codeowner.schema.get("rules") is None
+        ):
+            return
+
+        # Convert raw to issue owners syntax so that the schema can be created
+        raw = codeowner.raw
+        associations, _ = validate_codeowners_associations(codeowner.raw, project)
+        codeowner.raw = convert_codeowners_syntax(
+            codeowner.raw,
+            associations,
+            codeowner.repository_project_path_config,
+        )
+        codeowner.schema = create_schema_from_issue_owners(
+            project_id=project.id,
+            issue_owners=codeowner.raw,
+            add_owner_ids=True,
+            remove_deleted_owners=True,
+        )
+
+        # Convert raw back to codeowner type to be saved
+        codeowner.raw = raw
+
+        codeowner.save()
+
+    def get(self, request: Request, project: Project) -> Response:
+        """
+        Retrieve List of CODEOWNERS configurations for a project
+        ````````````````````````````````````````````
+
+        Return a list of a project's CODEOWNERS configuration.
+
+        :auth: required
+        """
+
+        if not self.has_feature(request, project):
+            raise PermissionDenied
+
+        expand = request.GET.getlist("expand", [])
+        expand.append("errors")
+
+        codeowners = list(ProjectCodeOwners.objects.filter(project=project).order_by("-date_added"))
+        for codeowner in codeowners:
+            self.refresh_codeowners_schema(codeowner, project)
+        expand.append("renameIdentifier")
+        expand.append("hasTargetingContext")
+
+        return Response(
+            serialize(
+                codeowners,
+                request.user,
+                serializer=projectcodeowners_serializers.ProjectCodeOwnersSerializer(expand=expand),
+            ),
+            status.HTTP_200_OK,
+        )
+
+    def post(self, request: Request, project: Project) -> Response:
+        """
+        Upload a CODEOWNERS for project
+        `````````````
+
+        :pparam string organization_id_or_slug: the id or slug of the organization.
+        :pparam string project_id_or_slug: the id or slug of the project to get.
+        :param string raw: the raw CODEOWNERS text
+        :param string codeMappingId: id of the RepositoryProjectPathConfig object
+        :auth: required
+        """
+        if not self.has_feature(request, project):
+            self.track_response_code("create", PermissionDenied.status_code)
+            raise PermissionDenied
+
+        serializer = ProjectCodeOwnerSerializer(context={"project": project}, data=request.data)
+
+        if serializer.is_valid():
+            project_codeowners = serializer.save()
+            self.track_response_code("create", status.HTTP_201_CREATED)
+            user_id = getattr(request.user, "id", None) or None
+            try:
+                analytics.record(
+                    CodeOwnersCreated(
+                        user_id=user_id,
+                        organization_id=project.organization_id,
+                        project_id=project.id,
+                        codeowners_id=project_codeowners.id,
+                    )
+                )
+            except Exception as e:
+                sentry_sdk.capture_exception(e)
+
+            expand = ["ownershipSyntax", "errors", "hasTargetingContext"]
+
+            return Response(
+                serialize(
+                    project_codeowners,
+                    request.user,
+                    serializer=projectcodeowners_serializers.ProjectCodeOwnersSerializer(
+                        expand=expand
+                    ),
+                ),
+                status=status.HTTP_201_CREATED,
+            )
+
+        self.track_response_code("create", status.HTTP_400_BAD_REQUEST)
+        return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST)