feat(backup): Enable export over RPC

Exports are already done on a sequential, per-model basis, going in
reverse dependency order. This change modifies that scheme to work
across an RPC boundary: code running on REGION instances (the only kind
of instance that may perform import/export operations) requests the
exports using either the local implementation (if the model is REGION as
well), or by sending an RPC call to the same method on the remote
CONTROL silo instance. Code executed in MONOLITH mode (aka, self-hosted
and single tenant instances) does all exporting locally.

The new `import_export_service` is thus a bit different from some of the
others in `hybrid_cloud`. Whereas existing services like `Access`
provide a means to get the same information using different logic
depending on which Silo they are located in (ex: using `Organization` or
`OrganizationMapping` as the source of truth), this service executes
identical logic in all silo kinds, and requires users to pick the
implementation themselves. That is, users must examine the model they
are about to request be exported, and make their own decision as to
which implementation, remote or local, to use. This is a bit unergonomic
compared to other `hybrid_cloud` services, but it is only used by one
rather unique endpoint (the import/export functions) for a very specific
purpose, making this API a bit easier to swallow.

Issue: getsentry/team-ospo#196

Author: azaslavsky

src/sentry/backup/dependencies.py

@@ -4,7 +4,7 @@
 from dataclasses import dataclass
 from enum import Enum, auto, unique
 from functools import lru_cache
-from typing import Dict, FrozenSet, NamedTuple, Optional, Set, Tuple, Type
+from typing import NamedTuple, Optional, Tuple, Type
 
 from django.db import models
 from django.db.models.fields.related import ForeignKey, OneToOneField
@@ -150,6 +150,9 @@ def get_possible_relocation_scopes(self) -> set[RelocationScope]:
             return self.model.get_possible_relocation_scopes()
         return set()
 
+    def get_dependencies_for_relocation(self) -> set[Type[models.base.Model]]:
+        return self.flatten().union(self.relocation_dependencies)
+
     def get_uniques_without_foreign_keys(self) -> list[frozenset[str]]:
         """
         Gets all unique sets (that is, either standalone fields that are marked `unique=True`, or
@@ -177,7 +180,7 @@ def get_uniques_without_foreign_keys(self) -> list[frozenset[str]]:
         return out
 
 
-def get_model_name(model: type[models.Model] | models.Model) -> NormalizedModelName:
+def get_model_name(model: Type[models.Model] | models.Model) -> NormalizedModelName:
     return NormalizedModelName(f"{model._meta.app_label}.{model._meta.object_name}")
 
 
@@ -217,7 +220,7 @@ def default(self, obj):
         return super().default(obj)
 
 
-class ImportKind(Enum):
+class ImportKind(str, Enum):
     """
     When importing a given model, we may create a new copy of it (`Inserted`), merely re-use an
     `Existing` copy that has the same already-used globally unique identifier (ex: `username` for
@@ -227,9 +230,9 @@ class ImportKind(Enum):
     if they are dealing with a new or re-used model.
     """
 
-    Inserted = auto()
-    Existing = auto()
-    Overwrite = auto()
+    Inserted = "Inserted"
+    Existing = "Existing"
+    Overwrite = "Overwrite"
 
 
 class PrimaryKeyMap:
@@ -244,7 +247,7 @@ class PrimaryKeyMap:
     keys are not supported!
     """
 
-    mapping: Dict[str, Dict[int, Tuple[int, ImportKind, Optional[str]]]]
+    mapping: dict[str, dict[int, Tuple[int, ImportKind, Optional[str]]]]
 
     def __init__(self):
         self.mapping = defaultdict(dict)
@@ -264,7 +267,7 @@ def get_pk(self, model_name: NormalizedModelName, old: int) -> Optional[int]:
 
         return entry[0]
 
-    def get_pks(self, model_name: NormalizedModelName) -> Set[int]:
+    def get_pks(self, model_name: NormalizedModelName) -> set[int]:
         """
         Get a list of all of the pks for a specific model.
         """
@@ -315,6 +318,31 @@ def insert(
 
         self.mapping[str(model_name)][old] = (new, kind, slug)
 
+    def extend(self, other: PrimaryKeyMap) -> None:
+        """
+        Insert all values from another map into this one, without mutating the original map.
+        """
+
+        for model_name_str, mappings in other.mapping.items():
+            for old_pk, new_entry in mappings.items():
+                self.mapping[model_name_str][old_pk] = new_entry
+
+    def partition(self, model_names: set[NormalizedModelName]) -> PrimaryKeyMap:
+        """
+        Create a new map with only the specified model kinds retained.
+        """
+
+        building = PrimaryKeyMap()
+        for model_name_str, mappings in self.mapping.items():
+            model_name = NormalizedModelName(model_name_str)
+            if model_name not in model_names:
+                continue
+
+            for old_pk, new_entry in mappings.items():
+                building.mapping[model_name_str][old_pk] = new_entry
+
+        return building
+
 
 # No arguments, so we lazily cache the result after the first calculation.
 @lru_cache(maxsize=1)
@@ -336,7 +364,7 @@ def dependencies() -> dict[NormalizedModelName, ModelRelations]:
     from sentry.models.team import Team
 
     # Process the list of models, and get the list of dependencies.
-    model_dependencies_dict: Dict[NormalizedModelName, ModelRelations] = {}
+    model_dependencies_dict: dict[NormalizedModelName, ModelRelations] = {}
     app_configs = apps.get_app_configs()
     models_from_names = {
         get_model_name(model): model
@@ -351,8 +379,8 @@ def dependencies() -> dict[NormalizedModelName, ModelRelations]:
         model_iterator = app_config.get_models()
 
         for model in model_iterator:
-            foreign_keys: Dict[str, ForeignField] = dict()
-            uniques: Set[FrozenSet[str]] = {
+            foreign_keys: dict[str, ForeignField] = dict()
+            uniques: set[frozenset[str]] = {
                 frozenset(combo) for combo in model._meta.unique_together
             }
 
@@ -485,7 +513,7 @@ def dependencies() -> dict[NormalizedModelName, ModelRelations]:
     # models non-dangling, then traversing from every other model to a (possible) root model
     # recursively. At this point there should be no circular reference chains, so if we encounter
     # them, fail immediately.
-    def resolve_dangling(seen: Set[NormalizedModelName], model_name: NormalizedModelName) -> bool:
+    def resolve_dangling(seen: set[NormalizedModelName], model_name: NormalizedModelName) -> bool:
         model_relations = model_dependencies_dict[model_name]
         model_name = get_model_name(model_relations.model)
         if model_name in seen:
@@ -557,7 +585,7 @@ def sorted_dependencies() -> list[Type[models.base.Model]]:
         changed = False
         while model_dependencies_remaining:
             model_deps = model_dependencies_remaining.pop()
-            deps = model_deps.flatten().union(model_deps.relocation_dependencies)
+            deps = model_deps.get_dependencies_for_relocation()
             model = model_deps.model
 
             # If all of the models in the dependency list are either already