hybrid-cloud-rpc

Compare original and translation side by side

🇺🇸

Original

English

🇨🇳

Translation

Chinese

Hybrid Cloud RPC Services

混合云RPC服务

This skill guides you through creating, modifying, and deprecating RPC services in Sentry's hybrid cloud architecture. RPC services enable cross-silo communication between the Control silo (user auth, billing, org management) and Region silos (project data, events, issues).

本技能将指导你在Sentry的混合云架构中创建、修改和弃用RPC服务。RPC服务可实现Control silo（用户认证、账单、组织管理）和Region silo（项目数据、事件、问题）之间的跨silo通信。

Critical Constraints

关键约束

NEVER use
from __future__ import annotations
in
service.py
or
model.py
files. The RPC framework reflects on type annotations at import time. Forward references break serialization silently.

ALL RPC method parameters must be keyword-only (use
*
in the signature).

ALL parameters and return types must have full type annotations — no string forward references.

ONLY serializable types are allowed:
int
,
str
,
bool
,
float
,
None
,
Optional[T]
,
list[T]
,
dict[str, T]
,
RpcModel
subclasses,
Enum
subclasses,
datetime.datetime
.

The service MUST live in one of the 12 registered discovery packages (see Step 3).

Use
Field(repr=False)
on sensitive fields (tokens, secrets, keys, config blobs, metadata dicts) to prevent them from leaking into logs and error reports. See
references/rpc-models.md
for the full guide.

严禁在
service.py
或
model.py
文件中使用
from __future__ import annotations
。 RPC框架会在导入时反射读取类型注解，前向引用会悄无声息地破坏序列化功能。

所有RPC方法参数必须是关键字参数（在签名中使用
*
分隔）。

所有参数和返回值必须有完整的类型注解——禁止使用字符串形式的前向引用。

仅允许使用可序列化类型：
int
、
str
、
bool
、
float
、
None
、
Optional[T]
、
list[T]
、
dict[str, T]
、
RpcModel
子类、
Enum
子类、
datetime.datetime
。

服务必须存放在12个已注册的发现包之一中（见步骤3）。

敏感字段（令牌、密钥、配置块、元数据字典）要使用
Field(repr=False)
，避免泄露到日志和错误报告中。完整指南请参考
references/rpc-models.md
。

Step 1: Determine Operation

步骤1：确定操作类型

Classify what the developer needs:

Intent	Go to
Create a brand-new RPC service	Step 2, then Step 3
Add a method to an existing service	Step 2, then Step 4
Update an existing method's signature	Step 5
Deprecate or remove a method/service	Step 6

对开发者的需求进行分类：

意图	跳转至
创建全新的RPC服务	步骤2，之后跳转步骤3
为现有服务添加方法	步骤2，之后跳转步骤4
更新现有方法的签名	步骤5
弃用或移除方法/服务	步骤6

Step 2: Determine Silo Mode

步骤2：确定Silo模式

The service's

local_mode

determines where the database-backed implementation runs:

Data lives in...	`local_mode`	Decorator on methods	Example
Region silo (projects, events, issues, org data)	`SiloMode.REGION`	`@regional_rpc_method(resolve=...)`	`OrganizationService`
Control silo (users, auth, billing, org mappings)	`SiloMode.CONTROL`	`@rpc_method`	`OrganizationMemberMappingService`

Decision rule: If the Django models you need to query live in the region database, use

SiloMode.REGION

. If they live in the control database, use

SiloMode.CONTROL

Region-silo services require a

RegionResolutionStrategy

on every RPC method so the framework knows which region to route remote calls to. Load

references/resolvers.md

for the full resolver table.

服务的

local_mode

决定了数据库侧的实现运行位置：

数据存储位置	`local_mode`	方法装饰器	示例
Region silo（项目、事件、问题、组织数据）	`SiloMode.REGION`	`@regional_rpc_method(resolve=...)`	`OrganizationService`
Control silo（用户、认证、账单、组织映射）	`SiloMode.CONTROL`	`@rpc_method`	`OrganizationMemberMappingService`

判定规则：如果你需要查询的Django模型存放在区域数据库中，使用

SiloMode.REGION

。如果存放在控制数据库中，使用

SiloMode.CONTROL

。

Region-silo服务的每个RPC方法都需要配置

RegionResolutionStrategy

，这样框架才能知道远程调用要路由到哪个区域。完整的解析器表请查看

references/resolvers.md

。

Step 3: Create a New Service

步骤3：创建新服务

Load

references/service-template.md

for copy-paste file templates.

可加载

references/service-template.md

获取可直接复制粘贴的文件模板。

Directory structure

目录结构

src/sentry/{domain}/services/{service_name}/
├── __init__.py      # Re-exports model and service
├── model.py         # RpcModel subclasses (NO future annotations)
├── serial.py        # ORM → RpcModel conversion functions
├── service.py       # Abstract service class (NO future annotations)
└── impl.py          # DatabaseBacked implementation

src/sentry/{domain}/services/{service_name}/
├── __init__.py      # 重导出模型和服务
├── model.py         # RpcModel子类（禁止使用future注解）
├── serial.py        # ORM → RpcModel转换函数
├── service.py       # 抽象服务类（禁止使用future注解）
└── impl.py          # 数据库侧实现类

Registration

注册要求

The service package MUST be a sub-package of one of these 12 registered discovery packages:

sentry.auth.services
sentry.audit_log.services
sentry.backup.services
sentry.hybridcloud.services
sentry.identity.services
sentry.integrations.services
sentry.issues.services
sentry.notifications.services
sentry.organizations.services
sentry.projects.services
sentry.sentry_apps.services
sentry.users.services

If your service doesn't fit any of these, add a new entry to the

service_packages

tuple in

src/sentry/hybridcloud/rpc/service.py:list_all_service_method_signatures()

服务包必须是以下12个已注册的发现包的子包：

sentry.auth.services
sentry.audit_log.services
sentry.backup.services
sentry.hybridcloud.services
sentry.identity.services
sentry.integrations.services
sentry.issues.services
sentry.notifications.services
sentry.organizations.services
sentry.projects.services
sentry.sentry_apps.services
sentry.users.services

如果你的服务不符合以上任何分类，需要在

src/sentry/hybridcloud/rpc/service.py:list_all_service_method_signatures()

的

service_packages

元组中添加新条目。

Checklist for new services

新服务检查清单

```
key
```
is unique across all services (check existing keys with
```
grep -r 'key = "' src/sentry/*/services/*/service.py
```
)
```
local_mode
```
matches where the data lives

get_local_implementation()

returns the

DatabaseBacked

subclass

Module-level

my_service = MyService.create_delegation()

at bottom of

service.py

```
__init__.py
```
re-exports models and service

from __future__ import annotations

service.py

model.py

key

在所有服务中唯一（可通过

grep -r 'key = "' src/sentry/*/services/*/service.py

检查现有key）

```
local_mode
```
与数据存储位置匹配

get_local_implementation()

DatabaseBacked

子类

在

service.py

底部的模块层级执行

my_service = MyService.create_delegation()

```
__init__.py
```
重导出模型和服务

service.py

和

model.py

中没有

from __future__ import annotations

Step 4: Add or Update Methods

步骤4：添加或更新方法

For REGION silo services

REGION silo服务配置

Load

references/resolvers.md

for resolver details.

python

@regional_rpc_method(resolve=ByOrganizationId())
@abstractmethod
def my_method(
    self,
    *,
    organization_id: int,
    name: str,
    options: RpcMyOptions | None = None,
) -> RpcMyResult | None:
    pass

Key rules:

```
@regional_rpc_method
```
MUST come before
```
@abstractmethod
```
The resolver parameter (e.g.,
```
organization_id
```
) MUST be in the method signature
Use
```
return_none_if_mapping_not_found=True
```
when the return type is
```
Optional
```
and a missing org mapping means "not found" rather than an error

加载

references/resolvers.md

查看解析器详细说明。

python

@regional_rpc_method(resolve=ByOrganizationId())
@abstractmethod
def my_method(
    self,
    *,
    organization_id: int,
    name: str,
    options: RpcMyOptions | None = None,
) -> RpcMyResult | None:
    pass

核心规则：

```
@regional_rpc_method
```
必须放在
```
@abstractmethod
```
之前
解析器参数（例如
```
organization_id
```
）必须出现在方法签名中
当返回类型是
```
Optional
```
且缺失组织映射代表“未找到”而非错误时，使用
```
return_none_if_mapping_not_found=True
```

For CONTROL silo services

CONTROL silo服务配置

python

@rpc_method
@abstractmethod
def my_method(
    self,
    *,
    user_id: int,
    data: RpcMyData,
) -> RpcMyResult:
    pass

python

@rpc_method
@abstractmethod
def my_method(
    self,
    *,
    user_id: int,
    data: RpcMyData,
) -> RpcMyResult:
    pass

Non-abstract convenience methods

非抽象便捷方法

You can also add non-abstract methods that compose other RPC calls. These run locally and are NOT exposed as RPC endpoints:

python

def get_by_slug_or_id(self, *, slug: str | None = None, id: int | None = None) -> RpcThing | None:
    if slug:
        return self.get_by_slug(slug=slug)
    if id:
        return self.get_by_id(id=id)
    return None

你也可以添加组合其他RPC调用的非抽象方法，这些方法仅在本地运行，不会暴露为RPC端点：

python

def get_by_slug_or_id(self, *, slug: str | None = None, id: int | None = None) -> RpcThing | None:
    if slug:
        return self.get_by_slug(slug=slug)
    if id:
        return self.get_by_id(id=id)
    return None

Implementation in impl.py

impl.py中的实现

The

DatabaseBacked

subclass must implement every

@abstractmethod

with the exact same parameter names:

python

class DatabaseBackedMyService(MyService):
    def my_method(self, *, organization_id: int, name: str, options: RpcMyOptions | None = None) -> RpcMyResult | None:
        # ORM queries here
        obj = MyModel.objects.filter(organization_id=organization_id, name=name).first()
        if obj is None:
            return None
        return serialize_my_model(obj)

DatabaseBacked

子类必须实现所有

@abstractmethod

，且参数名称完全一致：

python

class DatabaseBackedMyService(MyService):
    def my_method(self, *, organization_id: int, name: str, options: RpcMyOptions | None = None) -> RpcMyResult | None:
        # 此处编写ORM查询逻辑
        obj = MyModel.objects.filter(organization_id=organization_id, name=name).first()
        if obj is None:
            return None
        return serialize_my_model(obj)

Error propagation

错误传播

All errors an RPC method propagates must be done via the return type. Errors are rewrapped and returned as generic Invalid service request to external callers.

python

class RpcTentativeResult(RpcModel):
    success: bool
    error_str: str | None
    result: str | None

class DatabaseBackedMyService(MyService):
    def foobar(self, *, organization_id: int) -> RpcTentativeResult
        try:
            some_function_call()
        except e:
            return RpcTentativeResult(success=False, error_str = str(e))

        return RpcTentativeResult(success=True, result="foobar")

RPC方法抛出的所有错误都必须通过返回值处理，错误会被重新包装为通用的无效服务请求返回给外部调用方。

python

class RpcTentativeResult(RpcModel):
    success: bool
    error_str: str | None
    result: str | None

class DatabaseBackedMyService(MyService):
    def foobar(self, *, organization_id: int) -> RpcTentativeResult
        try:
            some_function_call()
        except e:
            return RpcTentativeResult(success=False, error_str = str(e))

        return RpcTentativeResult(success=True, result="foobar")

RPC Models

RPC模型

Load

references/rpc-models.md

for supported types, default values, and serialization patterns.

加载

references/rpc-models.md

查看支持的类型、默认值和序列化模式。

Step 5: Update Method Signatures

步骤5：更新方法签名

Safe changes (backwards compatible)

安全变更（向后兼容）

Adding a new optional parameter with a default value
Widening a return type (e.g.,
```
RpcFoo
```
→
```
RpcFoo | None
```
) on a Control RPC service
Adding fields with defaults to an
```
RpcModel
```

添加带默认值的新可选参数
在Control RPC服务上拓宽返回类型（例如
```
RpcFoo
```
→
```
RpcFoo | None
```
）
为
```
RpcModel
```
添加带默认值的字段

Breaking changes (require coordination)

破坏性变更（需要协调）

Removing or renaming a parameter
Changing a parameter's type
Narrowing a return type
Removing fields from an
```
RpcModel
```

For breaking changes, use a two-phase approach:

Add the new method alongside the old one
Migrate all callers to the new method
Remove the old method (see Step 6)

移除或重命名参数
变更参数类型
收窄返回类型
从
```
RpcModel
```
中移除字段

对于破坏性变更，采用两阶段方案：

与旧方法并行添加新方法
将所有调用方迁移到新方法
移除旧方法（见步骤6）

Step 6: Deprecate or Remove

步骤6：弃用或移除

Load

references/deprecation.md

for the full 3-phase workflow.

Quick summary: Disable at runtime → migrate callers → remove code.

加载

references/deprecation.md

查看完整的三阶段工作流。

快速概要：运行时禁用 → 迁移调用方 → 移除代码。

Step 7: Test

步骤7：测试

Every RPC service needs three categories of tests: silo mode compatibility, data accuracy, and error handling. Use

TransactionTestCase

(not

TestCase

) when tests need outbox processing or

on_commit

hooks.

每个RPC服务都需要三类测试：silo模式兼容性、数据准确性和错误处理。当测试需要outbox处理或

on_commit

钩子时，使用

TransactionTestCase

（而非

TestCase

）。

7.1 Silo mode compatibility with

@all_silo_test

7.1 使用

@all_silo_test

测试Silo模式兼容性

Every service test class MUST use

@all_silo_test

so tests run in all three modes (MONOLITH, REGION, CONTROL). This ensures the delegation layer works for both local and remote dispatch paths.

python

from sentry.testutils.cases import TestCase, TransactionTestCase
from sentry.testutils.silo import all_silo_test, assume_test_silo_mode, create_test_regions

@all_silo_test
class MyServiceTest(TestCase):
    def test_get_by_id(self):
        org = self.create_organization()
        result = my_service.get_by_id(organization_id=org.id, id=thing.id)
        assert result is not None

For tests that need named regions (e.g., testing region resolution):

python

@all_silo_test(regions=create_test_regions("us", "eu"))
class MyServiceRegionTest(TransactionTestCase):
    ...

Use

assume_test_silo_mode

assume_test_silo_mode_of

to switch modes within a test when accessing ORM models that live in a different silo:

python

def test_cross_silo_behavior(self):
    with assume_test_silo_mode(SiloMode.REGION):
        org = self.create_organization()
    result = my_service.get_by_id(organization_id=org.id, id=thing.id)
    assert result is not None

每个服务测试类都必须使用

@all_silo_test

，这样测试会在三种模式（MONOLITH、REGION、CONTROL）下都运行，确保代理层在本地和远程调度路径下都能正常工作。

python

from sentry.testutils.cases import TestCase, TransactionTestCase
from sentry.testutils.silo import all_silo_test, assume_test_silo_mode, create_test_regions

@all_silo_test
class MyServiceTest(TestCase):
    def test_get_by_id(self):
        org = self.create_organization()
        result = my_service.get_by_id(organization_id=org.id, id=thing.id)
        assert result is not None

对于需要指定区域的测试（例如测试区域解析）：

python

@all_silo_test(regions=create_test_regions("us", "eu"))
class MyServiceRegionTest(TransactionTestCase):
    ...

当访问存放在其他silo的ORM模型时，可使用

assume_test_silo_mode

或

assume_test_silo_mode_of

在测试中切换模式：

python

def test_cross_silo_behavior(self):
    with assume_test_silo_mode(SiloMode.REGION):
        org = self.create_organization()
    result = my_service.get_by_id(organization_id=org.id, id=thing.id)
    assert result is not None

7.2 Serialization round-trip with

dispatch_to_local_service

7.2 使用

dispatch_to_local_service

测试序列化往返

Test that arguments and return values survive serialization/deserialization:

python

from sentry.hybridcloud.rpc.service import dispatch_to_local_service

def test_serialization_round_trip(self):
    result = dispatch_to_local_service(
        "my_service_key",
        "my_method",
        {"organization_id": org.id, "name": "test"},
    )
    assert result["value"] is not None

测试参数和返回值可以正常完成序列化/反序列化：

python

from sentry.hybridcloud.rpc.service import dispatch_to_local_service

def test_serialization_round_trip(self):
    result = dispatch_to_local_service(
        "my_service_key",
        "my_method",
        {"organization_id": org.id, "name": "test"},
    )
    assert result["value"] is not None

7.3 RPC model data accuracy

7.3 RPC模型数据准确性

Validate that RPC models faithfully represent the ORM data. Compare every field of the RPC model against the source ORM object:

python

def test_rpc_model_accuracy(self):
    orm_obj = MyModel.objects.get(id=thing.id)
    rpc_obj = my_service.get_by_id(organization_id=org.id, id=thing.id)

    assert rpc_obj.id == orm_obj.id
    assert rpc_obj.name == orm_obj.name
    assert rpc_obj.organization_id == orm_obj.organization_id
    assert rpc_obj.is_active == orm_obj.is_active
    assert rpc_obj.date_added == orm_obj.date_added

For models with flags or nested objects, iterate all field names:

python

def test_flags_accuracy(self):
    rpc_org = organization_service.get(id=org.id)
    for field_name in rpc_org.flags.get_field_names():
        assert getattr(rpc_org.flags, field_name) == getattr(orm_org.flags, field_name)

For list results, sort both sides by ID before comparing:

python

def test_list_accuracy(self):
    rpc_items = my_service.list_things(organization_id=org.id)
    orm_items = list(MyModel.objects.filter(organization_id=org.id).order_by("id"))
    assert len(rpc_items) == len(orm_items)
    for rpc_item, orm_item in zip(sorted(rpc_items, key=lambda x: x.id), orm_items):
        assert rpc_item.id == orm_item.id
        assert rpc_item.name == orm_item.name

验证RPC模型可以准确表示ORM数据，将RPC模型的所有字段与源ORM对象对比：

python

def test_rpc_model_accuracy(self):
    orm_obj = MyModel.objects.get(id=thing.id)
    rpc_obj = my_service.get_by_id(organization_id=org.id, id=thing.id)

    assert rpc_obj.id == orm_obj.id
    assert rpc_obj.name == orm_obj.name
    assert rpc_obj.organization_id == orm_obj.organization_id
    assert rpc_obj.is_active == orm_obj.is_active
    assert rpc_obj.date_added == orm_obj.date_added

对于带标志或嵌套对象的模型，遍历所有字段名：

python

def test_flags_accuracy(self):
    rpc_org = organization_service.get(id=org.id)
    for field_name in rpc_org.flags.get_field_names():
        assert getattr(rpc_org.flags, field_name) == getattr(orm_org.flags, field_name)

对于列表返回结果，对比前先按ID对两边排序：

python

def test_list_accuracy(self):
    rpc_items = my_service.list_things(organization_id=org.id)
    orm_items = list(MyModel.objects.filter(organization_id=org.id).order_by("id"))
    assert len(rpc_items) == len(orm_items)
    for rpc_item, orm_item in zip(sorted(rpc_items, key=lambda x: x.id), orm_items):
        assert rpc_item.id == orm_item.id
        assert rpc_item.name == orm_item.name

7.4 Cross-silo resource creation

7.4 跨Silo资源创建

If your service creates or updates resources that propagate across silos (via outboxes or mappings), verify the cross-silo effects.

Use

outbox_runner()

to flush outboxes synchronously during tests:

python

from sentry.testutils.outbox import outbox_runner

def test_cross_silo_mapping_created(self):
    with outbox_runner():
        my_service.create_thing(organization_id=org.id, name="test")

    with assume_test_silo_mode(SiloMode.CONTROL):
        mapping = MyMapping.objects.get(organization_id=org.id)
        assert mapping.name == "test"

For triple-equality assertions (RPC result = source ORM = cross-silo replica):

python

def test_provisioning_accuracy(self):
    rpc_result = my_service.provision(organization_id=org.id, slug="test")
    with assume_test_silo_mode(SiloMode.REGION):
        orm_obj = MyModel.objects.get(id=rpc_result.id)
    with assume_test_silo_mode(SiloMode.CONTROL):
        mapping = MyMapping.objects.get(organization_id=org.id)
    assert rpc_result.slug == orm_obj.slug == mapping.slug

Use

HybridCloudTestMixin

for common cross-silo assertions:

python

from sentry.testutils.hybrid_cloud import HybridCloudTestMixin

class MyServiceTest(HybridCloudTestMixin, TransactionTestCase):
    def test_member_mapping_synced(self):
        self.assert_org_member_mapping(org_member=org_member)

如果你的服务会创建或更新跨silo传播的资源（通过outbox或映射），需要验证跨silo生效情况。

测试期间使用

outbox_runner()

同步刷新outbox：

python

from sentry.testutils.outbox import outbox_runner

def test_cross_silo_mapping_created(self):
    with outbox_runner():
        my_service.create_thing(organization_id=org.id, name="test")

    with assume_test_silo_mode(SiloMode.CONTROL):
        mapping = MyMapping.objects.get(organization_id=org.id)
        assert mapping.name == "test"

对于三方相等断言（RPC结果 = 源ORM = 跨silo副本）：

python

def test_provisioning_accuracy(self):
    rpc_result = my_service.provision(organization_id=org.id, slug="test")
    with assume_test_silo_mode(SiloMode.REGION):
        orm_obj = MyModel.objects.get(id=rpc_result.id)
    with assume_test_silo_mode(SiloMode.CONTROL):
        mapping = MyMapping.objects.get(organization_id=org.id)
    assert rpc_result.slug == orm_obj.slug == mapping.slug

使用

HybridCloudTestMixin

实现通用跨silo断言：

python

from sentry.testutils.hybrid_cloud import HybridCloudTestMixin

class MyServiceTest(HybridCloudTestMixin, TransactionTestCase):
    def test_member_mapping_synced(self):
        self.assert_org_member_mapping(org_member=org_member)

7.5 Error handling

7.5 错误处理

Test that the service handles errors correctly in all silo modes:

python

def test_not_found_returns_none(self):
    result = my_service.get_by_id(organization_id=org.id, id=99999)
    assert result is None

def test_missing_org_returns_none(self):
    # For methods with return_none_if_mapping_not_found=True
    result = my_service.get_by_id(organization_id=99999, id=1)
    assert result is None

Test disabled methods:

python

from sentry.hybridcloud.rpc.service import RpcDisabledException
from sentry.testutils.helpers.options import override_options

def test_disabled_method_raises(self):
    with override_options({"hybrid_cloud.rpc.disabled-service-methods": ["MyService.my_method"]}):
        with pytest.raises(RpcDisabledException):
            dispatch_remote_call(None, "my_service_key", "my_method", {"id": 1})

Test that remote exceptions are properly wrapped:

python

from sentry.hybridcloud.rpc.service import RpcRemoteException

def test_remote_error_wrapping(self):
    if SiloMode.get_current_mode() == SiloMode.REGION:
        with pytest.raises(RpcRemoteException):
            my_control_service.do_thing_that_fails(...)

Test that failed operations produce no side effects:

python

def test_no_side_effects_on_failure(self):
    result = my_service.create_conflicting_thing(organization_id=org.id)
    assert not result
    with assume_test_silo_mode(SiloMode.REGION):
        assert not MyModel.objects.filter(organization_id=org.id).exists()

Test that any calling code (both direct and indirect) is also appropriately tested with the correct silo decorators.

测试服务在所有silo模式下都能正确处理错误：

python

def test_not_found_returns_none(self):
    result = my_service.get_by_id(organization_id=org.id, id=99999)
    assert result is None

def test_missing_org_returns_none(self):
    # 适用于配置了return_none_if_mapping_not_found=True的方法
    result = my_service.get_by_id(organization_id=99999, id=1)
    assert result is None

测试已禁用的方法：

python

from sentry.hybridcloud.rpc.service import RpcDisabledException
from sentry.testutils.helpers.options import override_options

def test_disabled_method_raises(self):
    with override_options({"hybrid_cloud.rpc.disabled-service-methods": ["MyService.my_method"]}):
        with pytest.raises(RpcDisabledException):
            dispatch_remote_call(None, "my_service_key", "my_method", {"id": 1})

测试远程异常被正确包装：

python

from sentry.hybridcloud.rpc.service import RpcRemoteException

def test_remote_error_wrapping(self):
    if SiloMode.get_current_mode() == SiloMode.REGION:
        with pytest.raises(RpcRemoteException):
            my_control_service.do_thing_that_fails(...)

测试失败的操作不会产生副作用：

python

def test_no_side_effects_on_failure(self):
    result = my_service.create_conflicting_thing(organization_id=org.id)
    assert not result
    with assume_test_silo_mode(SiloMode.REGION):
        assert not MyModel.objects.filter(organization_id=org.id).exists()

同时要确保所有调用代码（直接和间接调用）都使用了正确的silo装饰器进行了适配测试。

7.6 Key imports for testing

7.6 测试常用导入

python

from sentry.testutils.cases import TestCase, TransactionTestCase
from sentry.testutils.silo import (
    all_silo_test,
    control_silo_test,
    region_silo_test,
    assume_test_silo_mode,
    assume_test_silo_mode_of,
    create_test_regions,
)
from sentry.testutils.outbox import outbox_runner
from sentry.testutils.hybrid_cloud import HybridCloudTestMixin
from sentry.hybridcloud.rpc.service import (
    dispatch_to_local_service,
    dispatch_remote_call,
    RpcDisabledException,
    RpcRemoteException,
)

python

from sentry.testutils.cases import TestCase, TransactionTestCase
from sentry.testutils.silo import (
    all_silo_test,
    control_silo_test,
    region_silo_test,
    assume_test_silo_mode,
    assume_test_silo_mode_of,
    create_test_regions,
)
from sentry.testutils.outbox import outbox_runner
from sentry.testutils.hybrid_cloud import HybridCloudTestMixin
from sentry.hybridcloud.rpc.service import (
    dispatch_to_local_service,
    dispatch_remote_call,
    RpcDisabledException,
    RpcRemoteException,
)

Step 8: Verify (Pre-flight Checklist)

步骤8：验证（上线前检查清单）

Before submitting your PR, verify:

No
```
from __future__ import annotations
```
in service.py or model.py
All RPC method parameters are keyword-only (
```
*
```
separator)
All parameters have explicit type annotations
All types are serializable (primitives, RpcModel, list, Optional, dict, Enum, datetime)
Region service methods have
```
@regional_rpc_method
```
with appropriate resolver
Control service methods have
```
@rpc_method
```

@regional_rpc_method

@rpc_method

comes BEFORE

@abstractmethod

```
create_delegation()
```
is called at module level at the bottom of service.py
Service package is under one of the 12 registered discovery packages
```
impl.py
```
implements every abstract method with matching parameter names
```
serial.py
```
correctly converts ORM models to RPC models
Sensitive fields use
```
Field(repr=False)
```
(tokens, secrets, config, metadata)
Tests use
```
@all_silo_test
```
for full silo mode coverage
Tests validate RPC model field accuracy against ORM objects
Tests verify cross-silo resources (mappings, replicas) are created with correct data
Tests cover error cases (not found, disabled methods, failed operations)
Tests cover serialization round-trip via
```
dispatch_to_local_service
```

提交PR前，验证以下内容：

service.py和model.py中没有
```
from __future__ import annotations
```
所有RPC方法参数都是关键字参数（使用
```
*
```
分隔）
所有参数都有显式类型注解
所有类型都是可序列化的（基础类型、RpcModel、list、Optional、dict、Enum、datetime）
区域服务方法配置了带合适解析器的
```
@regional_rpc_method
```
控制服务方法配置了
```
@rpc_method
```

@regional_rpc_method

@rpc_method

放在

@abstractmethod

之前

在service.py底部的模块层级调用了
```
create_delegation()
```
服务包位于12个已注册的发现包之下
```
impl.py
```
实现了所有抽象方法，且参数名称匹配
```
serial.py
```
可以正确将ORM模型转换为RPC模型
敏感字段使用了
```
Field(repr=False)
```
（令牌、密钥、配置、元数据）
测试使用了
```
@all_silo_test
```
实现全silo模式覆盖
测试验证了RPC模型字段与ORM对象的准确性
测试验证了跨silo资源（映射、副本）的数据正确性
测试覆盖了错误场景（未找到、方法禁用、操作失败）
测试通过
```
dispatch_to_local_service
```
覆盖了序列化往返场景

hybrid-cloud-rpc

Original

Translation

Hybrid Cloud RPC Services

混合云RPC服务

Critical Constraints

关键约束

Step 1: Determine Operation

步骤1：确定操作类型

Step 2: Determine Silo Mode

步骤2：确定Silo模式

Step 3: Create a New Service

步骤3：创建新服务

Directory structure

目录结构

Registration

注册要求

Checklist for new services

新服务检查清单

Step 4: Add or Update Methods

步骤4：添加或更新方法

For REGION silo services

REGION silo服务配置

For CONTROL silo services

CONTROL silo服务配置

Non-abstract convenience methods

非抽象便捷方法

Implementation in impl.py

impl.py中的实现

Error propagation

错误传播

RPC Models

RPC模型

Step 5: Update Method Signatures

步骤5：更新方法签名

Safe changes (backwards compatible)

安全变更（向后兼容）

Breaking changes (require coordination)

破坏性变更（需要协调）

Step 6: Deprecate or Remove

步骤6：弃用或移除

Step 7: Test

步骤7：测试

7.1 Silo mode compatibility with @all_silo_test

7.1 使用@all_silo_test测试Silo模式兼容性

7.2 Serialization round-trip with dispatch_to_local_service

7.2 使用dispatch_to_local_service测试序列化往返

7.3 RPC model data accuracy

7.3 RPC模型数据准确性

7.4 Cross-silo resource creation

7.4 跨Silo资源创建

7.5 Error handling

7.5 错误处理

7.6 Key imports for testing

7.6 测试常用导入

Step 8: Verify (Pre-flight Checklist)

步骤8：验证（上线前检查清单）

7.1 Silo mode compatibility with
`@all_silo_test`

7.1 使用
`@all_silo_test`
测试Silo模式兼容性

7.2 Serialization round-trip with
`dispatch_to_local_service`

7.2 使用
`dispatch_to_local_service`
测试序列化往返