fix(backup): Preserve multiple project keys on import

src/sentry/models/project.py

@@ -976,11 +976,12 @@ def normalize_before_relocation_import(
     def write_relocation_import(
         self, scope: ImportScope, flags: ImportFlags
     ) -> tuple[int, ImportKind] | None:
+        from sentry.receivers.core import disable_default_project_key_creation
         from sentry.workflow_engine.receivers.project_detectors import (
             disable_default_detector_creation,
         )
 
-        with disable_default_detector_creation():
+        with disable_default_detector_creation(), disable_default_project_key_creation():
             return super().write_relocation_import(scope, flags)
 
     # pending deletion implementation

src/sentry/models/projectkey.py

@@ -11,7 +11,6 @@
 from django.conf import settings
 from django.db import ProgrammingError, models
 from django.db.models.signals import pre_delete
-from django.forms import model_to_dict
 from django.urls import reverse
 from django.utils import timezone
 from django.utils.translation import gettext_lazy as _
@@ -393,16 +392,7 @@ def write_relocation_import(
         if not self.secret_key or matching_secret_key:
             self.secret_key = self.generate_api_key()
 
-        # ProjectKeys for the project are automatically generated at insertion time via a
-        # `post_save()` hook, so the keys for the project should already exist. We simply need to
-        # update them with the correct values here.
-        (key, _) = ProjectKey.objects.get_or_create(
-            project=self.project, defaults=model_to_dict(self)
-        )
-        if key:
-            self.pk = key.pk
-            self.save()
-
+        self.save(force_insert=True)
         return (self.pk, ImportKind.Inserted)
 
 

src/sentry/receivers/core.py

@@ -1,4 +1,7 @@
 import os
+from collections.abc import Iterator
+from contextlib import contextmanager
+from contextvars import ContextVar
 
 from click import echo
 from django.conf import settings
@@ -28,6 +31,19 @@
 """
 DEFAULT_SENTRY_PROJECT_ID = 1
 
+_default_project_key_creation_enabled = ContextVar(
+    "default_project_key_creation_enabled", default=True
+)
+
+
+@contextmanager
+def disable_default_project_key_creation() -> Iterator[None]:
+    token = _default_project_key_creation_enabled.set(False)
+    try:
+        yield
+    finally:
+        _default_project_key_creation_enabled.reset(token)
+
 
 def create_default_projects(**kwds):
     if not (in_test_environment() or is_self_hosted() or settings.DEBUG):
@@ -113,6 +129,9 @@ def create_default_project(id, name, slug, verbosity=2, **kwargs):
 
 
 def create_keys_for_project(instance, created, app=None, **kwargs):
+    if not _default_project_key_creation_enabled.get():
+        return
+
     if app and app.__name__ != "sentry.models":
         return
 

src/sentry/testutils/factories.py

@@ -849,7 +849,9 @@ def create_slack_project_rule(project, integration_id, channel_id=None, channel_
 
     @staticmethod
     @assume_test_silo_mode(SiloMode.CELL)
-    def create_project_key(project):
+    def create_project_key(project, **kwargs):
+        if kwargs:
+            return project.key_set.create(**kwargs)
         return project.key_set.get_or_create()[0]
 
     @staticmethod

tests/sentry/backup/test_imports.py

@@ -729,7 +729,14 @@ def test_import_signaling_organization(self) -> None:
         owner = self.create_exhaustive_user("owner")
         invited = self.create_exhaustive_user("invited")
         member = self.create_exhaustive_user("member")
-        self.create_exhaustive_organization("some-org", owner, invited, [member])
+        organization = self.create_exhaustive_organization("some-org", owner, invited, [member])
+        project = Project.objects.get(organization=organization)
+        self.create_project_key(
+            project=project,
+            label="Secondary",
+            public_key="a" * 32,
+            secret_key="b" * 32,
+        )
 
         with tempfile.TemporaryDirectory() as tmp_dir:
             tmp_path = self.export_to_tmp_file_and_clear_database(tmp_dir)
@@ -742,11 +749,17 @@ def test_import_signaling_organization(self) -> None:
 
         assert OrganizationMember.objects.count() == 3
 
-        # The exhaustive org has 1 project which automatically gets 1 key and 3 options.
+        # The exhaustive org has 1 project with 2 keys and 3 options.
         assert Project.objects.count() == 1
         assert Project.objects.filter(name="project-some-org").exists()
 
-        assert ProjectKey.objects.count() == 1
+        assert ProjectKey.objects.count() == 2
+        assert ProjectKey.objects.filter(
+            project=imported_organization.project_set.get(),
+            label="Secondary",
+            public_key="a" * 32,
+            secret_key="b" * 32,
+        ).exists()
         assert ProjectOption.objects.count() == 1
         assert ProjectOption.objects.filter(key="sentry:option-epoch").exists()
 

tests/sentry/runner/commands/test_backup.py

@@ -27,6 +27,7 @@
 from sentry.backup.findings import InstanceID
 from sentry.backup.imports import ImportingError
 from sentry.backup.services.import_export.model import RpcImportError, RpcImportErrorKind
+from sentry.models.projectkey import ProjectKey
 from sentry.runner.commands.backup import backup, export, import_
 from sentry.silo.base import SiloMode
 from sentry.testutils.cases import TestCase, TransactionTestCase
@@ -538,6 +539,47 @@ def test_global_scope(self) -> None:
         clear_database()
         cli_import_then_export("global")
 
+    def test_global_scope_imports_multiple_project_keys(self) -> None:
+        with open(GOOD_FILE_PATH, "rb") as fixture_file:
+            dump = orjson.loads(fixture_file.read())
+
+        project_key_index = next(
+            index for index, item in enumerate(dump) if item["model"] == "sentry.projectkey"
+        )
+        first_key = dump[project_key_index]
+        second_key = {
+            **first_key,
+            "pk": first_key["pk"] + 1,
+            "fields": {
+                **first_key["fields"],
+                "label": "Secondary",
+                "public_key": "a" * 32,
+                "secret_key": "b" * 32,
+            },
+        }
+        dump.insert(project_key_index + 1, second_key)
+
+        with TemporaryDirectory() as tmp_dir:
+            dump_path = Path(tmp_dir) / "multiple-project-keys.json"
+            dump_path.write_bytes(orjson.dumps(dump))
+
+            clear_database()
+            result = CliRunner().invoke(
+                import_, ["global", str(dump_path), "--no-prompt"], catch_exceptions=False
+            )
+
+        assert result.exit_code == 0, result.output
+        assert set(ProjectKey.objects.values_list("public_key", "secret_key")) == {
+            (
+                first_key["fields"]["public_key"],
+                first_key["fields"]["secret_key"],
+            ),
+            (
+                second_key["fields"]["public_key"],
+                second_key["fields"]["secret_key"],
+            ),
+        }
+
     def test_config_scope(self) -> None:
         cli_import_then_export("config")
         cli_import_then_export("config", import_args=["--silent"])