“

Series: S6 Platform Engineering · Article 14 | Level: Advanced | Reading Time: 18 min

38 gRPC Client Code Generation: From Protobuf to Production-Grade SDK

#TL;DR

coomia-dip spans 8 Layers with all internal communication built on gRPC. To ensure cross-language consistency and developer productivity, we built a unified code generation pipeline: starting from 38 .proto definitions, it automatically generates Java, Python, and TypeScript gRPC Client Stubs with production-grade features including connection pooling, retry policies, interceptor chains, and health checking. This article comprehensively covers the design philosophy, toolchain configuration, template engine, quality assurance, and CI/CD practices behind this code generation system.

#1. Why Unified gRPC Client Code Generation

#1.1 Challenges of Microservice Communication

In coomia-dip's 8-Layer architecture, inter-service communication is the nervous system. We face these core challenges:

Cross-language consistency: Control Layer + Data Layer use Java (Spring Boot / Quarkus), Reasoning & Decision Layer + Agent Runtime Layer use Python (FastAPI), and SDK & Developer Experience Layer's SDK must support both Python and TypeScript. Hand-writing Client code inevitably leads to protocol inconsistencies.
Maintenance cost: 38 service interfaces mean 38 × 3 = 114 Client codebases. Any .proto change requires synchronized updates across all languages—manual maintenance is unsustainable.
Production requirements: Raw gRPC Stubs cannot meet production needs; connection pooling, retries, timeouts, metrics collection, and distributed tracing are essential cross-cutting concerns.
Type safety: The platform is Ontology-centric with the type system spanning the full stack. Code generation must ensure precise type mappings.

#1.2 Design Philosophy Inspired by Palantir

Palantir Foundry internally uses a highly automated code generation system (Conjure) to generate multi-language Clients from API definitions. coomia-dip adopts this philosophy with an equivalent open-source approach:

Capability	Palantir Conjure	coomia-dip Codegen
IDL Definition	Conjure YAML	Protobuf 3
Transport Protocol	HTTP/2 + JSON	gRPC (HTTP/2 + Protobuf)
Java Client	Auto-generated	protoc + custom templates
Python Client	Auto-generated	grpcio-tools + wrapper generation
TypeScript Client	Auto-generated	grpc-web + ts-proto
Production enhancements	Built-in	Interceptor chain + pool + retry

#1.3 Distribution of 38 Services

coomia-dip's 38 gRPC services are distributed across Layers as follows:

Code

Control Layer (Control Layer)     — 12 services
├── OntologyService          ├── PolicyEngineService
├── ObjectTypeService        ├── ClassificationService
├── LinkTypeService          ├── DataMaskService
├── ActionTypeService        ├── WorkflowService
├── PropertyTypeService      ├── AuditService
├── ConstraintService        └── NotificationService

Data Layer (Data Layer)        — 8 services
├── DatasetService           ├── IcebergCatalogService
├── QueryService             ├── NessieBranchService
├── TransformService         ├── MaterializationService
├── PipelineService          └── DataQualityService

Reasoning & Decision Layer (Intelligence)      — 8 services
├── ReasoningService         ├── DecisionService
├── RuleEngineService        ├── SimulationService
├── ModelRegistryService     ├── FeatureStoreService
├── DerivedPropertyService   └── ScorecardService

Agent Runtime Layer (Agent Runtime)     — 6 services
├── AgentService             ├── ToolService
├── SessionService           ├── MemoryService
├── PlannerService           └── ExecutorService

Deployment & Operations Layer (Deployment)        — 4 services
├── DeploymentService        ├── ConfigService
├── HealthService            └── MetricsService

#2. Protobuf Definition Standards

#2.1 Directory Structure

All .proto files are centralized in the proto/ directory, organized by Layer and domain:

Code

proto/
├── common/
│   ├── pagination.proto      # Unified pagination
│   ├── error.proto           # Unified error codes
│   ├── metadata.proto        # Request metadata
│   └── types.proto           # Common types
├── plane_b/
│   ├── ontology/
│   │   ├── ontology_service.proto
│   │   ├── object_type_service.proto
│   │   └── link_type_service.proto
│   ├── security/
│   │   ├── policy_engine_service.proto
│   │   └── classification_service.proto
│   └── workflow/
│       └── workflow_service.proto
├── plane_c/
│   ├── dataset/
│   │   └── dataset_service.proto
│   └── query/
│       └── query_service.proto
├── plane_d/
│   └── reasoning/
│       └── reasoning_service.proto
└── plane_e/
    └── agent/
        └── agent_service.proto

#2.2 Naming and Style Conventions

PROTOBUF

syntax = "proto3";

package onto.plane_b.ontology.v1;

option java_package = "com.onto.plane_b.ontology.v1";
option java_multiple_files = true;
option go_package = "github.com/coomia-dip/proto/plane_b/ontology/v1";

// Service definition: Verb + Noun
service OntologyService {
  // Single resource operations
  rpc GetObjectType(GetObjectTypeRequest)
      returns (GetObjectTypeResponse);

  // List operations: unified pagination
  rpc ListObjectTypes(ListObjectTypesRequest)
      returns (ListObjectTypesResponse);

  // Create operations
  rpc CreateObjectType(CreateObjectTypeRequest)
      returns (CreateObjectTypeResponse);

  // Streaming operations
  rpc StreamChanges(StreamChangesRequest)
      returns (stream ChangeEvent);
}

// Message naming: {Method}Request / {Method}Response
message GetObjectTypeRequest {
  string object_type_rid = 1;  // Resource IDs use RID suffix
  FieldMask field_mask = 2;    // Field mask for partial responses
}

#2.3 Unified Conventions

We established strict Protobuf authoring standards to ensure generated code consistency:

Package naming: onto.{Layer}.{domain}.v{version}
Service naming: {Domain}Service
Method naming: {Verb}{Resource}, verbs are Get/List/Create/Update/Delete/Stream/Batch
Message naming: {MethodName}Request / {MethodName}Response
Field numbers: 1-15 for high-frequency fields (single-byte encoding), 16+ for low-frequency
Enums: First value must be UNSPECIFIED = 0
Timestamps: Always use google.protobuf.Timestamp
Pagination: Always use common.PaginationRequest / common.PaginationResponse

#3. Code Generation Pipeline Architecture

#3.1 Three-Stage Pipeline

Code generation follows a three-stage pipeline design:

Code

Stage 1: Proto Compilation   Stage 2: Enhancement Layer   Stage 3: Package & Publish
┌─────────────────┐      ┌─────────────────┐      ┌─────────────────┐
│  .proto files   │      │  Raw Stubs +    │      │  Language-specific│
│       │         │      │  Templates      │      │  Packages        │
│       ▼         │      │       │         │      │       │          │
│  protoc         │─────▶│  Template Engine │─────▶│  Maven Central  │
│  + plugins      │      │  (Jinja2/FreeMarker)   │  PyPI            │
│       │         │      │       │         │      │  npm registry    │
│       ▼         │      │       ▼         │      │                  │
│  Raw Stubs      │      │  Enhanced Clients│      │  Versioned       │
│  (Java/Py/TS)   │      │  + Interceptors │      │  artifacts       │
└─────────────────┘      └─────────────────┘      └─────────────────┘

#3.2 Stage 1: Protobuf Compilation

The first stage uses the buf toolchain instead of native protoc for better linting and breaking change detection:

YAML

# buf.yaml
version: v1
name: buf.build/coomia-dip/proto
lint:
  use:
    - DEFAULT
    - COMMENTS
  except:
    - PACKAGE_VERSION_SUFFIX
breaking:
  use:
    - FILE

YAML

# buf.gen.yaml
version: v1
plugins:
  # Java
  - plugin: buf.build/protocolbuffers/java
    out: gen/java
  - plugin: buf.build/grpc/java
    out: gen/java

  # Python
  - plugin: buf.build/protocolbuffers/python
    out: gen/python
  - plugin: buf.build/grpc/python
    out: gen/python
  - plugin: buf.build/protocolbuffers/pyi
    out: gen/python

  # TypeScript
  - plugin: buf.build/community/stephenh-ts-proto
    out: gen/typescript
    opt:
      - outputServices=grpc-js
      - esModuleInterop=true
      - useOptionals=messages
      - snakeToCamel=true

#3.3 Stage 2: Enhancement Layer Generation

Raw Stubs lack production-grade features. Enhancement layer templates add the following capabilities to each Client:

Python

# Python enhancement layer template (Jinja2)
class {{ service_name }}Client:
    """
    {{ service_description }}

    Production-grade gRPC Client, auto-generated.
    Includes connection pooling, retry, timeout, metrics collection.
    """

    def __init__(
        self,
        host: str = "localhost",
        port: int = 50051,
        *,
        timeout: float = 30.0,
        max_retries: int = 3,
        pool_size: int = 5,
        interceptors: list[ClientInterceptor] | None = None,
        credentials: grpc.ChannelCredentials | None = None,
    ):
        self._pool = ChannelPool(
            host=host, port=port,
            size=pool_size,
            credentials=credentials,
        )
        self._timeout = timeout
        self._retry_policy = RetryPolicy(
            max_retries=max_retries,
            retryable_codes=[
                grpc.StatusCode.UNAVAILABLE,
                grpc.StatusCode.DEADLINE_EXCEEDED,
            ],
        )
        self._interceptors = interceptors or [
            MetricsInterceptor(),
            TracingInterceptor(),
            LoggingInterceptor(),
        ]

    {% for method in methods %}
    def {{ method.python_name }}(
        self,
        request: {{ method.request_type }},
        *,
        timeout: float | None = None,
        metadata: dict[str, str] | None = None,
    ) -> {{ method.response_type }}:
        """{{ method.description }}"""
        return self._invoke(
            method="{{ method.full_name }}",
            request=request,
            response_type={{ method.response_type }},
            timeout=timeout or self._timeout,
            metadata=metadata,
        )
    {% endfor %}

#3.4 Stage 3: Package and Publish

Each language's Client packages are independently versioned and published:

Code

# Java: Gradle publish to internal Maven repository
onto-grpc-clients-java/
├── build.gradle.kts
├── src/main/java/com/onto/client/
│   ├── OntologyServiceClient.java
│   ├── DatasetServiceClient.java
│   └── ... (38 clients)
└── src/main/java/com/onto/client/common/
    ├── ChannelPool.java
    ├── RetryInterceptor.java
    └── TracingInterceptor.java

# Python: Publish to internal PyPI
onto-grpc-clients/
├── pyproject.toml
├── onto_grpc_clients/
│   ├── __init__.py
│   ├── py.typed
│   ├── ontology_client.py
│   └── ... (38 clients)

# TypeScript: Publish to internal npm
@onto/grpc-clients/
├── package.json
├── src/
│   ├── ontologyClient.ts
│   └── ... (38 clients)

#4. Connection Pooling and Channel Management

#4.1 Connection Pool Design

gRPC uses HTTP/2 multiplexing, allowing a single connection to carry multiple concurrent RPCs. However, under high load, single-connection bandwidth limits and head-of-line blocking persist. Our connection pool uses a weighted round-robin strategy:

Python

class ChannelPool:
    """gRPC connection pool with weighted round-robin and health checking."""

    def __init__(
        self,
        host: str,
        port: int,
        size: int = 5,
        credentials: grpc.ChannelCredentials | None = None,
    ):
        self._channels: list[grpc.Channel] = []
        self._index = 0
        self._lock = threading.Lock()
        self._health_check_interval = 30  # seconds

        options = [
            ("grpc.keepalive_time_ms", 10000),
            ("grpc.keepalive_timeout_ms", 5000),
            ("grpc.keepalive_permit_without_calls", True),
            ("grpc.http2.max_pings_without_data", 0),
            ("grpc.max_receive_message_length", 100 * 1024 * 1024),
        ]

        for _ in range(size):
            if credentials:
                ch = grpc.secure_channel(
                    f"{host}:{port}", credentials, options=options
                )
            else:
                ch = grpc.insecure_channel(
                    f"{host}:{port}", options=options
                )
            self._channels.append(ch)

    def get_channel(self) -> grpc.Channel:
        """Round-robin channel selection."""
        with self._lock:
            ch = self._channels[self._index % len(self._channels)]
            self._index += 1
            return ch

    async def health_check(self) -> dict[int, bool]:
        """Check health status of all connections."""
        results = {}
        for i, ch in enumerate(self._channels):
            try:
                state = ch.get_state(try_to_connect=True)
                results[i] = state == grpc.ChannelConnectivity.READY
            except Exception:
                results[i] = False
        return results

#4.2 Channel Lifecycle

Code

Creation                   Active                    Shutdown
┌──────┐   health_check   ┌──────┐   close          ┌──────┐
│ IDLE │ ──────────────▶  │READY │ ──────────────▶  │CLOSED│
└──────┘                  └──────┘                  └──────┘
    │                         │
    │    TRANSIENT_FAILURE     │    Connection lost
    └────────────────────────▶└──────────────────┐
                                                  ▼
                                             ┌──────────┐
                                             │CONNECTING│
                                             │(auto-reconnect)│
                                             └──────────┘

#5. Retry Policies and Error Handling

#5.1 Tiered Retry Policy

Different types of RPC calls require different retry strategies:

Python

class RetryPolicy:
    """Tiered retry policy based on method idempotency."""

    # Idempotent methods: safe to retry
    IDEMPOTENT_METHODS = {"Get", "List", "Search", "Count", "Exists"}

    # Non-idempotent methods: retry only on specific codes
    NON_IDEMPOTENT_RETRYABLE_CODES = {
        grpc.StatusCode.UNAVAILABLE,
    }

    # Idempotent methods: broader retry scope
    IDEMPOTENT_RETRYABLE_CODES = {
        grpc.StatusCode.UNAVAILABLE,
        grpc.StatusCode.DEADLINE_EXCEEDED,
        grpc.StatusCode.ABORTED,
        grpc.StatusCode.RESOURCE_EXHAUSTED,
    }

    def __init__(
        self,
        max_retries: int = 3,
        base_delay: float = 0.1,
        max_delay: float = 30.0,
        backoff_multiplier: float = 2.0,
    ):
        self.max_retries = max_retries
        self.base_delay = base_delay
        self.max_delay = max_delay
        self.backoff_multiplier = backoff_multiplier

    def should_retry(
        self,
        method_name: str,
        error: grpc.RpcError,
        attempt: int,
    ) -> tuple[bool, float]:
        """Determine whether to retry, returns (should_retry, wait_time)."""
        if attempt >= self.max_retries:
            return False, 0

        code = error.code()
        is_idempotent = any(
            method_name.startswith(prefix)
            for prefix in self.IDEMPOTENT_METHODS
        )

        retryable_codes = (
            self.IDEMPOTENT_RETRYABLE_CODES
            if is_idempotent
            else self.NON_IDEMPOTENT_RETRYABLE_CODES
        )

        if code not in retryable_codes:
            return False, 0

        delay = min(
            self.base_delay * (self.backoff_multiplier ** attempt),
            self.max_delay,
        )
        # Add jitter
        jitter = delay * 0.2 * (2 * random.random() - 1)
        return True, delay + jitter

#5.2 Unified Error Mapping

Unified mapping between gRPC status codes and business error types:

Python

ERROR_MAPPING = {
    grpc.StatusCode.NOT_FOUND: ResourceNotFoundError,
    grpc.StatusCode.ALREADY_EXISTS: ResourceConflictError,
    grpc.StatusCode.PERMISSION_DENIED: PermissionDeniedError,
    grpc.StatusCode.UNAUTHENTICATED: AuthenticationError,
    grpc.StatusCode.INVALID_ARGUMENT: ValidationError,
    grpc.StatusCode.FAILED_PRECONDITION: PreconditionError,
    grpc.StatusCode.RESOURCE_EXHAUSTED: RateLimitError,
    grpc.StatusCode.INTERNAL: InternalError,
    grpc.StatusCode.UNAVAILABLE: ServiceUnavailableError,
    grpc.StatusCode.DEADLINE_EXCEEDED: TimeoutError,
}

def translate_error(error: grpc.RpcError) -> OntoError:
    """Translate gRPC errors to business exceptions."""
    code = error.code()
    details = error.details() or "Unknown error"

    error_class = ERROR_MAPPING.get(code, InternalError)

    # Attempt to extract structured error info from trailing metadata
    metadata = dict(error.trailing_metadata() or [])
    error_id = metadata.get("x-error-id", "")
    error_context = metadata.get("x-error-context", "")

    return error_class(
        message=details,
        error_id=error_id,
        context=error_context,
        grpc_code=code,
    )

#6. Interceptor Chain

#6.1 Interceptor Architecture

The interceptor chain provides unified handling of cross-cutting concerns for every RPC call:

Code

Request ──▶ [Auth] ──▶ [Tracing] ──▶ [Metrics] ──▶ [Logging] ──▶ [Retry] ──▶ Server
                                                                                │
Response ◀── [Auth] ◀── [Tracing] ◀── [Metrics] ◀── [Logging] ◀── [Retry] ◀──┘

#6.2 Core Interceptor Implementations

Python

class TracingInterceptor(grpc.UnaryUnaryClientInterceptor):
    """OpenTelemetry distributed tracing interceptor."""

    def __init__(self, tracer_name: str = "onto-grpc-client"):
        self._tracer = trace.get_tracer(tracer_name)

    def intercept_unary_unary(self, continuation, client_call_details, request):
        method = client_call_details.method.decode("utf-8")
        with self._tracer.start_as_current_span(
            name=f"grpc.{method}",
            kind=trace.SpanKind.CLIENT,
            attributes={
                "rpc.system": "grpc",
                "rpc.method": method,
                "rpc.service": method.rsplit("/", 1)[0],
            },
        ) as span:
            # Inject trace context into metadata
            metadata = list(client_call_details.metadata or [])
            inject(carrier=metadata, setter=GrpcMetadataSetter())

            new_details = _ClientCallDetails(
                method=client_call_details.method,
                timeout=client_call_details.timeout,
                metadata=metadata,
                credentials=client_call_details.credentials,
            )

            try:
                response = continuation(new_details, request)
                span.set_status(trace.StatusCode.OK)
                return response
            except grpc.RpcError as e:
                span.set_status(
                    trace.StatusCode.ERROR,
                    description=str(e.details()),
                )
                span.set_attribute("rpc.grpc.status_code", e.code().value[0])
                raise


class MetricsInterceptor(grpc.UnaryUnaryClientInterceptor):
    """Prometheus metrics collection interceptor."""

    def __init__(self):
        self._request_counter = Counter(
            "grpc_client_requests_total",
            "Total gRPC client requests",
            ["method", "status"],
        )
        self._request_duration = Histogram(
            "grpc_client_request_duration_seconds",
            "gRPC client request duration",
            ["method"],
            buckets=[0.01, 0.05, 0.1, 0.5, 1.0, 5.0, 30.0],
        )

    def intercept_unary_unary(self, continuation, client_call_details, request):
        method = client_call_details.method.decode("utf-8")
        start_time = time.monotonic()

        try:
            response = continuation(client_call_details, request)
            self._request_counter.labels(
                method=method, status="OK"
            ).inc()
            return response
        except grpc.RpcError as e:
            self._request_counter.labels(
                method=method, status=e.code().name
            ).inc()
            raise
        finally:
            duration = time.monotonic() - start_time
            self._request_duration.labels(method=method).observe(duration)

#7. Type Safety in Generated Code

#7.1 Python Type Mapping

Protobuf-to-Python type mapping requires special attention for Pydantic v2 compatibility:

Protobuf Type	Python Type	Pydantic Field
string	str	Field(...)
int32/int64	int	Field(...)
float/double	float	Field(...)
bool	bool	Field(default=False)
bytes	bytes	Field(...)
Timestamp	datetime	Field(...)
repeated T	list[T]	Field(default_factory=list)
map<K,V>	dict[K,V]	Field(default_factory=dict)
oneof	Union[...]	Field(discriminator=...)
enum	IntEnum	Field(...)

#7.2 Conversion Layer Generation

To avoid business code directly depending on Protobuf-generated classes, we generate a conversion layer between Pydantic models and Protobuf:

Python

class ObjectTypeModel(BaseModel):
    """Pydantic model for Ontology ObjectType."""
    model_config = ConfigDict(frozen=True)

    rid: str
    api_name: str
    display_name: str
    description: str = ""
    properties: list[PropertyTypeModel] = Field(default_factory=list)
    created_at: datetime | None = None
    updated_at: datetime | None = None

    @classmethod
    def from_proto(cls, proto: ObjectTypePb) -> "ObjectTypeModel":
        return cls(
            rid=proto.rid,
            api_name=proto.api_name,
            display_name=proto.display_name,
            description=proto.description,
            properties=[
                PropertyTypeModel.from_proto(p) for p in proto.properties
            ],
            created_at=(
                proto.created_at.ToDatetime()
                if proto.HasField("created_at") else None
            ),
            updated_at=(
                proto.updated_at.ToDatetime()
                if proto.HasField("updated_at") else None
            ),
        )

    def to_proto(self) -> ObjectTypePb:
        proto = ObjectTypePb(
            rid=self.rid,
            api_name=self.api_name,
            display_name=self.display_name,
            description=self.description,
        )
        for p in self.properties:
            proto.properties.append(p.to_proto())
        if self.created_at:
            proto.created_at.FromDatetime(self.created_at)
        if self.updated_at:
            proto.updated_at.FromDatetime(self.updated_at)
        return proto

#8. CI/CD Integration

#8.1 Breaking Change Detection

Every PR modifying .proto files automatically triggers breaking change detection in the CI pipeline:

YAML

# .gitlab-ci.yml
proto-lint:
  stage: validate
  script:
    - buf lint proto/
    - buf breaking proto/ --against .git#branch=main
  rules:
    - changes:
        - "proto/**/*.proto"

proto-generate:
  stage: generate
  script:
    - buf generate proto/
    - python scripts/generate_enhanced_clients.py
    - ./gradlew :grpc-clients:compileJava
    - cd gen/python && pip install -e . && mypy .
    - cd gen/typescript && npm install && tsc --noEmit
  artifacts:
    paths:
      - gen/

#8.2 Version Management Strategy

gRPC Client packages use semantic versioning, managed independently from Protobuf API versions:

Code

Proto API version: v1, v2 (in package path)
Client package version: independent SemVer

Version matrix:
┌────────────────┬──────────────┬──────────────┐
│ Proto API      │ Client Pkg   │ Compatibility │
├────────────────┼──────────────┼──────────────┤
│ v1             │ 1.0.0-1.x.x │ Full compat   │
│ v1 (new field) │ 1.x+1.0     │ Backward      │
│ v2 (breaking)  │ 2.0.0        │ Migration req │
└────────────────┴──────────────┴──────────────┘

#9. Java Client Generation Specifics

#9.1 Spring Boot Integration

Java Clients need seamless integration with Spring Boot 3.x dependency injection:

Java

@Configuration
@EnableConfigurationProperties(GrpcClientProperties.class)
public class GrpcClientAutoConfiguration {

    @Bean
    @ConditionalOnMissingBean
    public OntologyServiceClient ontologyServiceClient(
            GrpcClientProperties properties,
            List<ClientInterceptor> interceptors) {
        return new OntologyServiceClient(
            properties.getHost("ontology-service"),
            properties.getPort("ontology-service"),
            GrpcClientOptions.builder()
                .timeout(properties.getTimeout())
                .maxRetries(properties.getMaxRetries())
                .poolSize(properties.getPoolSize())
                .interceptors(interceptors)
                .build()
        );
    }

    // ... other 37 Client Bean definitions (via code generation)
}

#9.2 Quarkus Integration

Data Layer uses Quarkus, requiring CDI-compatible Client generation:

Java

@ApplicationScoped
public class DatasetServiceClientProducer {

    @ConfigProperty(name = "grpc.dataset-service.host")
    String host;

    @ConfigProperty(name = "grpc.dataset-service.port")
    int port;

    @Produces
    @ApplicationScoped
    public DatasetServiceClient datasetServiceClient() {
        return new DatasetServiceClient(host, port);
    }
}

#10. Performance Optimization

#10.1 Connection Pool Tuning

Based on production load testing data, we compiled a connection pool tuning guide:

Scenario	Recommended Pool Size	Keepalive	Max Message Size
Low load (<100 QPS)	2	30s	4MB
Medium load (100-1000 QPS)	5	15s	16MB
High load (>1000 QPS)	10	10s	64MB
Large messages (Dataset transfers)	3	30s	100MB

#10.2 Message Compression

For large message bodies, enabling gzip compression significantly reduces network transfer:

Python

# Configure compression during Client initialization
channel_options = [
    ("grpc.default_compression_algorithm", grpc.Compression.Gzip),
    ("grpc.default_compression_level", grpc.CompressionLevel.Medium),
]

#10.3 Streaming RPC Optimization

For bulk data operations, use Server Streaming or Bidirectional Streaming instead of Unary RPCs:

Python

async def stream_query_results(
    self,
    request: StreamQueryRequest,
) -> AsyncIterator[QueryResult]:
    """Stream query results to avoid memory pressure from large result sets."""
    async with self._pool.get_channel() as channel:
        stub = QueryServiceStub(channel)
        async for response in stub.StreamQueryResults(request):
            yield QueryResult.from_proto(response)

#11. Testing Strategy

#11.1 Testing Generated Code

Code generation itself needs testing to ensure correctness:

Python

class TestCodeGeneration:
    """Test code generation pipeline correctness."""

    def test_all_services_generated(self):
        """Verify all 38 services have generated Clients."""
        generated_clients = list(Path("gen/python").glob("*_client.py"))
        assert len(generated_clients) == 38

    def test_type_mapping_completeness(self):
        """Verify all Protobuf types have corresponding Python type mappings."""
        for proto_file in Path("proto").rglob("*.proto"):
            types = extract_types(proto_file)
            for t in types:
                assert t in TYPE_MAPPING, f"Missing mapping for {t}"

    def test_generated_code_type_checks(self):
        """Run mypy type checking on generated code."""
        result = subprocess.run(
            ["mypy", "gen/python/"],
            capture_output=True, text=True,
        )
        assert result.returncode == 0, result.stderr

#11.2 Mock Server Testing

Use the grpc_testing library to create Mock Servers for integration testing:

Python

@pytest.fixture
def mock_ontology_server():
    """Create a Mock OntologyService Server."""
    servicers = {
        OntologyServiceServicer: MockOntologyServicer(),
    }
    server = grpc_testing.server_from_dictionary(
        servicers,
        grpc_testing.strict_real_time(),
    )
    yield server


def test_get_object_type(mock_ontology_server):
    request = GetObjectTypeRequest(object_type_rid="ri.ontology.object.1")
    method = mock_ontology_server.invoke_unary_unary(
        OntologyServiceServicer.GetObjectType,
        (),
        request,
        None,
    )
    response, metadata, code, details = method.termination()
    assert code == grpc.StatusCode.OK
    assert response.object_type.rid == "ri.ontology.object.1"

#12. Best Practices and Lessons Learned

#12.1 Proto Design Lessons

Pitfalls we encountered in practice:

Avoid deep nesting: Messages nested beyond 3 levels result in poor generated code readability
Reserve field numbers: Use reserved when removing fields, never delete directly
Limit batch sizes: BatchCreate and similar interfaces must have max_batch_size configuration
Add heartbeats to streams: Send heartbeats during idle periods to prevent connection timeouts

#12.2 Client Usage Best Practices

Python

# Correct: use context managers
async with OntologyServiceClient(host="control-Layer") as client:
    result = await client.get_object_type(request)

# Wrong: forgetting to close connections
client = OntologyServiceClient(host="control-Layer")
result = client.get_object_type(request)
# Connection leak!

# Correct: set reasonable timeouts
result = await client.get_object_type(request, timeout=5.0)

# Wrong: using default infinite timeout
result = await client.get_object_type(request)

#12.3 Version Compatibility Checklist

Before releasing a new Client version, execute the following checks:

buf breaking passes with no unexpected breaking changes
Generated code for all three languages compiles successfully
Type checking passes (mypy / tsc)
Unit tests pass
Mock Server integration tests pass
Version number follows SemVer
CHANGELOG has been updated

#Key Takeaways

Unified code generation is the fundamental guarantee for cross-language microservice communication consistency—maintenance cost for 38 services across 3 languages drops to zero incremental cost
Three-stage pipeline (compilation, enhancement, publishing) transforms raw Stubs into production-grade Clients with connection pooling, retry, tracing, and other cross-cutting concerns
buf toolchain replaces native protoc with lint and breaking change detection capabilities, representing best practices for Proto management
Interceptor chains achieve separation of concerns—each cross-cutting capability is independently developed, tested, and flexibly composed
Type mapping and conversion layers ensure business code is decoupled from Protobuf-generated code while maintaining type safety

#Next Article

Next, we will dive deep into TypeScript OSDK design and implementation, exploring how to provide frontend developers with a type-safe Ontology SDK.

Tags: gRPC Protobuf code-generation SDK microservices connection-pool interceptors buf platform-engineering