feat(data-secrecy): Create Data Secrecy RPC Service (#75694)

`sentry` version for getsentry/getsentry#14776

src/sentry/data_secrecy/service/impl.py

@@ -0,0 +1,16 @@
+from sentry.data_secrecy.models.datasecrecywaiver import DataSecrecyWaiver
+from sentry.data_secrecy.service.model import RpcDataSecrecyWaiver
+from sentry.data_secrecy.service.serial import serialize_data_secrecy_waiver
+from sentry.data_secrecy.service.service import DataSecrecyService
+
+
+class DatabaseBackedDataSecrecyService(DataSecrecyService):
+    def get_data_secrecy_waiver(self, *, organization_id: int) -> RpcDataSecrecyWaiver | None:
+        try:
+            data_secrecy_waiver = DataSecrecyWaiver.objects.filter(
+                organization_id=organization_id
+            ).get()
+        except DataSecrecyWaiver.DoesNotExist:
+            return None
+
+        return serialize_data_secrecy_waiver(data_secrecy_waiver=data_secrecy_waiver)

src/sentry/data_secrecy/service/model.py

@@ -0,0 +1,13 @@
+from datetime import datetime
+
+from django.utils import timezone
+from pydantic import Field
+
+from sentry.hybridcloud.rpc import RpcModel
+
+
+class RpcDataSecrecyWaiver(RpcModel):
+    organization_id: int = -1
+    access_start: datetime = Field(default_factory=timezone.now)
+    access_end: datetime = Field(default_factory=timezone.now)
+    zendesk_tickets: list[str] = Field(default_factory=list)

src/sentry/data_secrecy/service/serial.py

@@ -0,0 +1,11 @@
+from sentry.data_secrecy.models.datasecrecywaiver import DataSecrecyWaiver
+from sentry.data_secrecy.service.model import RpcDataSecrecyWaiver
+
+
+def serialize_data_secrecy_waiver(data_secrecy_waiver: DataSecrecyWaiver) -> RpcDataSecrecyWaiver:
+    return RpcDataSecrecyWaiver(
+        organization_id=data_secrecy_waiver.organization.id,
+        access_start=data_secrecy_waiver.access_start,
+        access_end=data_secrecy_waiver.access_end,
+        zendesk_tickets=data_secrecy_waiver.zendesk_tickets,
+    )

src/sentry/data_secrecy/service/service.py

@@ -0,0 +1,30 @@
+# Please do not use
+#     from __future__ import annotations
+# in modules such as this one where hybrid cloud data models or service classes are
+# defined, because we want to reflect on type annotations and avoid forward references.
+
+import abc
+
+from sentry.data_secrecy.service.model import RpcDataSecrecyWaiver
+from sentry.hybridcloud.rpc.resolvers import ByOrganizationId
+from sentry.hybridcloud.rpc.service import RpcService, regional_rpc_method
+from sentry.silo.base import SiloMode
+
+
+class DataSecrecyService(RpcService):
+    key = "data_secrecy"
+    local_mode = SiloMode.REGION
+
+    @classmethod
+    def get_local_implementation(cls) -> RpcService:
+        from sentry.data_secrecy.service.impl import DatabaseBackedDataSecrecyService
+
+        return DatabaseBackedDataSecrecyService()
+
+    @regional_rpc_method(resolve=ByOrganizationId())
+    @abc.abstractmethod
+    def get_data_secrecy_waiver(self, *, organization_id: int) -> RpcDataSecrecyWaiver | None:
+        pass
+
+
+data_secrecy_service = DataSecrecyService.create_delegation()

tests/sentry/data_secrecy/service/test_data_secrecy_service.py

@@ -0,0 +1,48 @@
+from datetime import datetime, timedelta, timezone
+
+from sentry.data_secrecy.models.datasecrecywaiver import DataSecrecyWaiver
+from sentry.data_secrecy.service.service import data_secrecy_service
+from sentry.silo.base import SiloMode
+from sentry.testutils.cases import TestCase
+from sentry.testutils.silo import all_silo_test, assume_test_silo_mode, create_test_regions
+
+
+@all_silo_test(regions=create_test_regions("us"))
+class TestDataSecrecyService(TestCase):
+    def setUp(self):
+        self.organization = self.create_organization()
+        self.organization_2 = self.create_organization()
+        self.access_start = datetime.now(tz=timezone.utc)
+        self.access_end = datetime.now(tz=timezone.utc) + timedelta(days=1)
+        self.zendesk_tickets = ["https://example.com/ticket1", "https://example.com/ticket2"]
+
+    def test_get_data_secrecy_waiver(self):
+        with assume_test_silo_mode(SiloMode.REGION):
+            self.waiver = DataSecrecyWaiver.objects.create(
+                organization_id=self.organization.id,
+                access_start=self.access_start,
+                access_end=self.access_end,
+                zendesk_tickets=self.zendesk_tickets,
+            )
+
+        # Test retrieving an existing waiver
+        result = data_secrecy_service.get_data_secrecy_waiver(organization_id=self.organization.id)
+        assert result is not None
+        assert result.organization_id == self.organization.id
+        assert result.access_start == self.access_start
+        assert result.access_end == self.access_end
+        assert result.zendesk_tickets == self.zendesk_tickets
+
+        # Test retrieving a non-existent waiver
+        non_existent_result = data_secrecy_service.get_data_secrecy_waiver(
+            organization_id=self.organization_2.id
+        )
+        assert non_existent_result is None
+
+        # Test after deleting the waiver
+        with assume_test_silo_mode(SiloMode.REGION):
+            self.waiver.delete()
+        deleted_result = data_secrecy_service.get_data_secrecy_waiver(
+            organization_id=self.organization.id
+        )
+        assert deleted_result is None