Skip to content

Commit

Permalink
chore(hybridcloud) Add non-tuple storage to RpcPrimaryKeyMap (#71439)
Browse files Browse the repository at this point in the history
In order to have RPC validation apply to all methods we need to not have
any tuples in method parameters/returns. Replace the tuple inside
RpcPrimaryKey with structured data. This will result in some redundant
data for a day or so until the tuples are removed.

Refs HC-1190
  • Loading branch information
markstory authored and volokluev committed May 28, 2024
1 parent e03e411 commit 49c5da9
Showing 1 changed file with 28 additions and 1 deletion.
29 changes: 28 additions & 1 deletion src/sentry/services/hybrid_cloud/import_export/model.py
Original file line number Diff line number Diff line change
Expand Up @@ -52,6 +52,12 @@ def into_rpc(cls, base_filter: Filter) -> "RpcFilter":
)


class RpcPrimaryKeyEntry(RpcModel):
new_id: int
kind: ImportKind
slug: str | None


class RpcPrimaryKeyMap(RpcModel):
"""
Shadows `sentry.backup.dependencies.PrimaryKeyMap` for the purpose of passing it over an RPC
Expand All @@ -62,17 +68,38 @@ class RpcPrimaryKeyMap(RpcModel):
"""

# Pydantic duplicates global default models on a per-instance basis, so using `{}` here is safe.
# Deprecated, use map_entries going forward as that property does not contain tuples.
mapping: dict[str, dict[int, tuple[int, ImportKind, str | None]]] = {}

# Pydantic duplicates global default models on a per-instance basis, so using `{}` here is safe.
map_entries: dict[str, dict[int, RpcPrimaryKeyEntry]] = {}

def from_rpc(self) -> PrimaryKeyMap:
pk_map = PrimaryKeyMap()
pk_map.mapping = defaultdict(dict, self.mapping)
if self.map_entries:
pk_mapping: dict[str, dict[int, tuple[int, ImportKind, str | None]]] = defaultdict(dict)
for model_name, entries in self.map_entries.items():
entries_data = dict()
for old_id, entry in entries.items():
entries_data[old_id] = (entry.new_id, entry.kind, entry.slug)
pk_mapping[model_name] = entries_data
pk_map.mapping = pk_mapping
else:
pk_map.mapping = defaultdict(dict, self.mapping)
return pk_map

@classmethod
def into_rpc(cls, base_map: PrimaryKeyMap) -> "RpcPrimaryKeyMap":
converted = cls()
converted.mapping = dict(base_map.mapping)

mapping_entries = dict()
for model_name, entries in base_map.mapping.items():
mapping_entries[model_name] = {
old_id: RpcPrimaryKeyEntry(new_id=entry[0], kind=entry[1], slug=entry[2])
for old_id, entry in entries.items()
}
converted.map_entries = mapping_entries
return converted


Expand Down

0 comments on commit 49c5da9

Please sign in to comment.