Hybrid Cloud Rpc

Hybrid Cloud RPC Services

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, org management) and Cell silos (project data, events, issues, billing).

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.

Step 1: Determine Operation

Classify what the developer needs:

Step 2: Determine Silo Mode

The service's local_mode determines where the database-backed implementation runs:

Decision rule: If the Django models you need to query live in the cell database, use SiloMode.CELL. If they live in the control database, use SiloMode.CONTROL.

Cell-silo services require a CellResolutionStrategy on every RPC method so the framework knows which cell to route remote calls to. Load references/resolvers.md for the full resolver table.

Step 3: Create a New Service

Load references/service-template.md for copy-paste file templates.

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

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().

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
• [ ] No from __future__ import annotations in service.py or model.py

Step 4: Add or Update Methods

For CELL silo services

Load references/resolvers.md for resolver details.

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

Key rules:

• @cell_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

For CONTROL silo services

@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:

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

The DatabaseBacked subclass must implement every @abstractmethod with the exact same parameter names:

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)

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.

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

Load references/rpc-models.md for supported types, default values, and serialization patterns.

Step 5: Update Method Signatures

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

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:

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

Step 6: Deprecate or Remove

Load references/deprecation.md for the full 3-phase workflow.

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

Step 7: Test

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.

7.1 Silo mode compatibility with @all_silo_test

Every service test class MUST use @all_silo_test so tests run in all three modes (MONOLITH, CELL, CONTROL). This ensures the delegation layer works for both local and remote dispatch paths.

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

@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 cells (e.g., testing cell resolution):

@all_silo_test(cells=create_test_cells("us", "eu"))
class MyServiceCellTest(TransactionTestCase):
    ...

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

def test_cross_silo_behavior(self):
    with assume_test_silo_mode(SiloMode.CELL):
        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

Test that arguments and return values survive serialization/deserialization:

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

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

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:

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:

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

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:

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):

def test_provisioning_accuracy(self):
    rpc_result = my_service.provision(organization_id=org.id, slug="test")
    with assume_test_silo_mode(SiloMode.CELL):
        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:

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

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

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:

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:

from sentry.hybridcloud.rpc.service import RpcRemoteException

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

Test that failed operations produce no side effects:

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.CELL):
        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.

7.6 Key imports for testing

from sentry.testutils.cases import TestCase, TransactionTestCase
from sentry.testutils.silo import (
    all_silo_test,
    control_silo_test,
    cell_silo_test,
    assume_test_silo_mode,
    assume_test_silo_mode_of,
    create_test_cells,
)
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)

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)
• [ ] Cell service methods have @cell_rpc_method with appropriate resolver
• [ ] Control service methods have @rpc_method
• [ ] @cell_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

Deprecating and Removing RPC Methods

Removing an RPC method or service requires a 3-phase approach to avoid breaking cross-silo calls during deployment windows where silos may be running different code versions.

Phase 1: Disable at Runtime

Use the hybrid_cloud.rpc.disabled-service-methods option to disable the method without removing code. This allows instant rollback.

# In Sentry options (or via the admin panel):
# Add "ServiceName.method_name" to the disabled list
options.set("hybrid_cloud.rpc.disabled-service-methods", [
    "MyService.old_method",
])

When a disabled method is called remotely, it raises RpcDisabledException instead of making a network call. This is checked in _RemoteSiloCall._check_disabled().

Runtime control options

Tip: Before removing a method, set its retries to 0 and timeout to a low value to surface any remaining callers.

Phase 2: Migrate Callers

1. Search for all callers of the method:

   grep -r "my_service\.old_method" src/ tests/

2. Update each caller to use the replacement method or inline the logic.

3. If the method is still needed during the transition, keep it implemented but add a deprecation comment:

   @cell_rpc_method(resolve=ByOrganizationId())
   @abstractmethod
   def old_method(self, *, organization_id: int) -> RpcThing | None:
       """Deprecated: Use new_method instead. Remove after YYYY-MM-DD."""
       pass

4. Deploy and verify zero traffic to the disabled method via metrics:

• Check hybrid_cloud.dispatch_rpc.response_code with tag rpc_method=ServiceName.old_method
• Confirm no RpcDisabledException errors in Sentry

Phase 3: Remove Code

Once confirmed that no callers remain:

1. Remove the abstract method from service.py 2. Remove the implementation from impl.py 3. Remove any RpcModel classes only used by the removed method from model.py 4. Remove serialization helpers from serial.py if now unused 5. Remove the method from the hybrid_cloud.rpc.disabled-service-methods option 6. Remove any related tests

Removing an entire service

If removing all methods from a service:

1. Complete Phase 1-3 for every method 2. Remove the create_delegation() call 3. Remove the service class 4. Remove the entire service directory 5. The service will automatically disappear from the registry since list_all_service_method_signatures() discovers via pkgutil.walk_packages

Safe Removal Checklist

• [ ] Method disabled via hybrid_cloud.rpc.disabled-service-methods in production
• [ ] Zero traffic confirmed via metrics for at least 1 week
• [ ] All callers migrated (grep confirms no references)
• [ ] Tests updated or removed
• [ ] RpcModel classes cleaned up if no longer used
• [ ] Option entry removed from disabled list after code removal

Cell Resolution Strategies

Cell-silo services (local_mode = SiloMode.CELL) require every RPC method to declare how to resolve the target cell from the method's arguments. This is done via the resolve parameter on @cell_rpc_method.

All resolvers are defined in src/sentry/hybridcloud/rpc/resolvers.py.

Resolver Table

Usage Examples

ByOrganizationId (most common)

# Uses default parameter_name="organization_id"
@cell_rpc_method(resolve=ByOrganizationId())
@abstractmethod
def get_thing(self, *, organization_id: int, id: int) -> RpcThing | None:
    pass

# Custom parameter name
@cell_rpc_method(resolve=ByOrganizationId("id"))
@abstractmethod
def serialize_organization(self, *, id: int) -> Any | None:
    pass

ByOrganizationSlug

# Uses default parameter_name="slug"
@cell_rpc_method(resolve=ByOrganizationSlug())
@abstractmethod
def get_org_by_slug(self, *, slug: str) -> RpcOrgSummary | None:
    pass

ByOrganizationIdAttribute

Resolves the cell from an attribute of an RpcModel parameter. Useful when the organization ID is embedded in a request object.

# parameter_name is required — it names the method parameter
# attribute_name defaults to "organization_id"
@cell_rpc_method(resolve=ByOrganizationIdAttribute("organization_member"))
@abstractmethod
def update_membership_flags(self, *, organization_member: RpcOrganizationMember) -> None:
    pass

In this example, the framework calls arguments["organization_member"].organization_id to get the org ID for cell lookup.

To use a different attribute:

@cell_rpc_method(
    resolve=ByOrganizationIdAttribute("request", attribute_name="org_id")
)
@abstractmethod
def process_request(self, *, request: RpcMyRequest) -> None:
    pass

ByCellName

# Uses default parameter_name="cell_name"
@cell_rpc_method(resolve=ByCellName())
@abstractmethod
def update_cell_user(self, *, user: RpcCellUser, cell_name: str) -> None:
    pass

RequireSingleOrganization

@cell_rpc_method(resolve=RequireSingleOrganization())
@abstractmethod
def get_default_organization(self) -> RpcOrganization:
    pass

This resolver raises CellResolutionError if the environment is not configured for single-organization mode.

return_none_if_mapping_not_found

When a method has an Optional return type and a missing organization mapping means "not found" rather than an error, set this flag:

@cell_rpc_method(resolve=ByOrganizationId("id"), return_none_if_mapping_not_found=True)
@abstractmethod
def get_organization_by_id(self, *, id: int) -> RpcOrganization | None:
    pass

Without this flag, a missing OrganizationMapping raises CellMappingNotFound. With it, the method returns None instead.

Use when: The method is a lookup that should gracefully handle deleted/unmapped orgs. Do NOT use when: A missing mapping indicates a bug or data integrity issue.

Choosing a Resolver

Decision tree:

1. Does the method take organization_id: int directly? → ByOrganizationId() 2. Does the method take slug: str? → ByOrganizationSlug() 3. Does the method take an RpcModel that has organization_id? → ByOrganizationIdAttribute("param_name") 4. Does the caller already know the cell name? → ByCellName() 5. Is it a single-org-only operation? → RequireSingleOrganization()

RPC Models

RPC models are Pydantic BaseModel subclasses inheriting from RpcModel (defined in src/sentry/hybridcloud/rpc/__init__.py). They are the serialization contract between silos.

Supported Types

Unsupported (will break serialization)

set, frozenset, tuple, bytes, Django model instances, Any, forward reference strings, types from from __future__ import annotations.

Model Template

# 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 datetime
from typing import Any

from pydantic import Field

from sentry.hybridcloud.rpc import DEFAULT_DATE, RpcModel


class RpcMyThing(RpcModel):
    id: int = -1
    name: str = ""
    tags: list[str] = Field(default_factory=list)
    config: dict[str, Any] = Field(repr=False, default_factory=dict)
    secret_token: str = Field(repr=False, default="")
    date_added: datetime.datetime = DEFAULT_DATE
    parent_id: int | None = None

    class Config:
        orm_mode = True
        use_enum_values = True

Hiding Sensitive Fields with repr=False

Use Field(repr=False) on any field that contains sensitive, bulky, or opaque data. This prevents the value from appearing in repr() output, log messages, Sentry breadcrumbs, and tracebacks.

When to use repr=False

Real examples from the codebase

# Tokens and secrets — always hide
token_hashed: str = Field(repr=False, default="")          # orgauthtoken/model.py
client_id: str = Field(repr=False, default="")              # app/model.py
client_secret: str = Field(repr=False, default="")          # app/model.py
session_nonce: str | None = Field(repr=False, default=None) # user/model.py
hash: str = Field(repr=False, default="")                   # lost_password_hash/model.py
token: str = Field(repr=False, default="")                  # auth/model.py
key: str = Field(repr=False, default="")                    # auth/model.py

# Opaque config/metadata — hide to reduce noise and avoid hidden PII
config: Any = Field(repr=False, default=None)               # user/model.py
metadata: dict[str, Any] = Field(repr=False)                # integration/model.py
extra_data: dict[str, Any] = Field(repr=False)              # usersocialauth/model.py
config: dict[str, Any] = Field(repr=False)                  # repository/model.py

# Request/response data — may contain auth headers
request_headers: Mapping[str, str] | None = Field(repr=False, default=None)

Syntax

from pydantic import Field

# With a default value
secret: str = Field(repr=False, default="")

# With a factory default
metadata: dict[str, Any] = Field(repr=False, default_factory=dict)

# Without a default (required field) — less common in RPC models
external_id: str = Field(repr=False)

Rule of thumb

If the field name contains any of these words, use repr=False: token, secret, key, hash, password, nonce, credential, config, metadata, extra_data, headers.

Serialization Methods

serialize_by_field_name (preferred)

Automatically maps fields by name from the source object:

def serialize_thing(obj: Thing) -> RpcThing:
    return RpcThing.serialize_by_field_name(obj)

With name_transform

When ORM field names differ from RPC field names:

RpcThing.serialize_by_field_name(
    obj,
    name_transform=lambda n: f"org_{n}" if n == "id" else n,
)

With value_transform

When values need conversion:

RpcThing.serialize_by_field_name(
    obj,
    value_transform=lambda v: v.value if isinstance(v, Enum) else v,
)

Manual construction

When the mapping is complex or involves computed fields:

def serialize_thing(obj: Thing) -> RpcThing:
    return RpcThing(
        id=obj.id,
        name=obj.name,
        is_active=obj.flags.is_active,
    )

Common Pitfalls

1. Missing defaults: Every RpcModel field MUST have a default. Pydantic requires this for deserialization when fields are added later.

2. `from __future__ import annotations`: Converts all annotations to strings, breaking Pydantic reflection. Never use in model files.

3. Mutable defaults: Use Field(default_factory=list) not = [] for list/dict defaults.

4. Enum fields: Declare field type as int (or str) and set use_enum_values = True in Config. Use the enum member as the default.

5. Nested RpcModels: Automatically serialized. Ensure nested models also have proper defaults.

6. Read vs Write models: Use separate models for read responses (RpcMyThing) and write payloads (RpcMyThingUpdate). Write models contain only mutable fields.

7. Forgetting `repr=False`: Sensitive fields without repr=False will appear in logs and error reports. Always audit new fields for sensitivity.

RPC Service File Templates

__init__.py

from .model import *  # noqa
from .service import *  # noqa

model.py (CELL silo example)

# 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 datetime
from typing import Any

from pydantic import Field

from sentry.hybridcloud.rpc import DEFAULT_DATE, RpcModel


class RpcMyThing(RpcModel):
    id: int = 0
    organization_id: int = 0
    name: str = ""
    is_active: bool = True
    # Use repr=False for opaque blobs and sensitive data to prevent log leakage
    config: dict[str, Any] = Field(repr=False, default_factory=dict)
    date_added: datetime.datetime = DEFAULT_DATE

    class Config:
        orm_mode = True
        use_enum_values = True


class RpcMyThingUpdate(RpcModel):
    """Write model for updates — only include mutable fields."""
    name: str = ""
    is_active: bool = True

model.py (CONTROL silo example)

# 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 datetime

from pydantic import Field

from sentry.hybridcloud.rpc import DEFAULT_DATE, RpcModel


class RpcMyMapping(RpcModel):
    id: int = 0
    user_id: int | None = None
    organization_id: int = 0
    role: str = ""
    email: str | None = None
    # Use repr=False for tokens, secrets, and credentials
    token: str = Field(repr=False, default="")
    date_added: datetime.datetime = DEFAULT_DATE

    class Config:
        orm_mode = True
        use_enum_values = True

serial.py

from __future__ import annotations

from sentry.myapp.models import MyThing
from sentry.mydomain.services.mything.model import RpcMyThing


def serialize_my_thing(obj: MyThing) -> RpcMyThing:
    return RpcMyThing.serialize_by_field_name(obj)

With name_transform

Use when the ORM field name differs from the RPC model field name:

def serialize_my_thing(obj: MyThing) -> RpcMyThing:
    return RpcMyThing.serialize_by_field_name(
        obj,
        name_transform=lambda n: f"thing_{n}" if n == "id" else n,
    )

With value_transform

Use when values need conversion (e.g., enum to int):

def serialize_my_thing(obj: MyThing) -> RpcMyThing:
    return RpcMyThing.serialize_by_field_name(
        obj,
        value_transform=lambda v: v.value if hasattr(v, "value") else v,
    )

Manual serialization

Use when the mapping is complex:

def serialize_my_thing(obj: MyThing) -> RpcMyThing:
    return RpcMyThing(
        id=obj.id,
        organization_id=obj.organization_id,
        name=obj.name,
        is_active=obj.flags.is_active,
        date_added=obj.date_added,
    )

service.py (CELL silo)

# 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.
from abc import abstractmethod

from sentry.hybridcloud.rpc.resolvers import ByOrganizationId
from sentry.hybridcloud.rpc.service import RpcService, cell_rpc_method
from sentry.mydomain.services.mything.model import RpcMyThing, RpcMyThingUpdate
from sentry.silo.base import SiloMode


class MyThingService(RpcService):
    key = "my_thing"
    local_mode = SiloMode.CELL

    @classmethod
    def get_local_implementation(cls) -> RpcService:
        from sentry.mydomain.services.mything.impl import DatabaseBackedMyThingService

        return DatabaseBackedMyThingService()

    @cell_rpc_method(resolve=ByOrganizationId())
    @abstractmethod
    def get_by_id(
        self,
        *,
        organization_id: int,
        id: int,
    ) -> RpcMyThing | None:
        pass

    @cell_rpc_method(resolve=ByOrganizationId())
    @abstractmethod
    def update(
        self,
        *,
        organization_id: int,
        id: int,
        attrs: RpcMyThingUpdate,
    ) -> RpcMyThing | None:
        pass


my_thing_service = MyThingService.create_delegation()

service.py (CONTROL silo)

# 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.
from abc import abstractmethod

from sentry.hybridcloud.rpc.service import RpcService, rpc_method
from sentry.mydomain.services.mymapping.model import RpcMyMapping
from sentry.silo.base import SiloMode


class MyMappingService(RpcService):
    key = "my_mapping"
    local_mode = SiloMode.CONTROL

    @classmethod
    def get_local_implementation(cls) -> RpcService:
        from sentry.mydomain.services.mymapping.impl import DatabaseBackedMyMappingService

        return DatabaseBackedMyMappingService()

    @rpc_method
    @abstractmethod
    def get_by_user(
        self,
        *,
        user_id: int,
    ) -> list[RpcMyMapping]:
        pass

    @rpc_method
    @abstractmethod
    def upsert(
        self,
        *,
        organization_id: int,
        user_id: int | None = None,
        role: str = "",
    ) -> RpcMyMapping:
        pass


my_mapping_service = MyMappingService.create_delegation()

impl.py (CELL silo example)

from __future__ import annotations

from sentry.myapp.models import MyThing
from sentry.mydomain.services.mything.model import RpcMyThing, RpcMyThingUpdate
from sentry.mydomain.services.mything.serial import serialize_my_thing
from sentry.mydomain.services.mything.service import MyThingService


class DatabaseBackedMyThingService(MyThingService):
    def get_by_id(
        self,
        *,
        organization_id: int,
        id: int,
    ) -> RpcMyThing | None:
        try:
            obj = MyThing.objects.get(organization_id=organization_id, id=id)
        except MyThing.DoesNotExist:
            return None
        return serialize_my_thing(obj)

    def update(
        self,
        *,
        organization_id: int,
        id: int,
        attrs: RpcMyThingUpdate,
    ) -> RpcMyThing | None:
        try:
            obj = MyThing.objects.get(organization_id=organization_id, id=id)
        except MyThing.DoesNotExist:
            return None
        obj.name = attrs.name
        obj.is_active = attrs.is_active
        obj.save()
        return serialize_my_thing(obj)

impl.py (CONTROL silo example)

from __future__ import annotations

from django.db import IntegrityError, router, transaction

from sentry.myapp.models import MyMapping
from sentry.mydomain.services.mymapping.model import RpcMyMapping
from sentry.mydomain.services.mymapping.serial import serialize_my_mapping
from sentry.mydomain.services.mymapping.service import MyMappingService


class DatabaseBackedMyMappingService(MyMappingService):
    def get_by_user(
        self,
        *,
        user_id: int,
    ) -> list[RpcMyMapping]:
        return [
            serialize_my_mapping(m)
            for m in MyMapping.objects.filter(user_id=user_id)
        ]

    def upsert(
        self,
        *,
        organization_id: int,
        user_id: int | None = None,
        role: str = "",
    ) -> RpcMyMapping:
        with transaction.atomic(router.db_for_write(MyMapping)):
            try:
                mapping, _ = MyMapping.objects.update_or_create(
                    organization_id=organization_id,
                    user_id=user_id,
                    defaults={"role": role},
                )
            except IntegrityError:
                mapping = MyMapping.objects.get(
                    organization_id=organization_id,
                    user_id=user_id,
                )
        return serialize_my_mapping(mapping)

Registered Discovery Packages

The following 12 packages are scanned for RPC services at startup (defined in src/sentry/hybridcloud/rpc/service.py:list_all_service_method_signatures()):

1. sentry.auth.services 2. sentry.audit_log.services 3. sentry.backup.services 4. sentry.hybridcloud.services 5. sentry.identity.services 6. sentry.integrations.services 7. sentry.issues.services 8. sentry.notifications.services 9. sentry.organizations.services 10. sentry.projects.services 11. sentry.sentry_apps.services 12. sentry.users.services

Your service's Python package must be a sub-package of one of these for automatic discovery.