spakky-grpc

gRPC 서비스 컨트롤러 통합 — code-first, 타입 안전 프로토콜 생성

스테레오타입

gRPC controller stereotype for service grouping.

Provides the @GrpcController stereotype for marking classes as gRPC service controllers with automatic service registration and protobuf package configuration.

GrpcController(package, service_name=None, *, name='', scope=Scope.SINGLETON) dataclass

Bases: Controller

Stereotype for gRPC service controllers.

Marks a class as a gRPC service controller with automatic service registration. Methods decorated with @rpc will be registered as gRPC service methods.

Attributes:

Name	Type	Description
package	str	Protobuf package name for the service.
service_name	str | None	gRPC service name. Defaults to the class name if not provided.

package instance-attribute

Protobuf package name for the service.

service_name = None class-attribute instance-attribute

gRPC service name. Defaults to the class name.

__call__(obj)

Apply the gRPC controller stereotype to a class.

Automatically generates the service name from the class name if not provided.

Parameters:

Name	Type	Description	Default
obj	type[T]	The class to decorate.	required

Returns:

Type	Description
type[T]	The decorated class registered as a Pod.

Source code in plugins/spakky-grpc/src/spakky/plugins/grpc/stereotypes/grpc_controller.py

def __call__[T: object](self, obj: type[T]) -> type[T]:
    """Apply the gRPC controller stereotype to a class.

    Automatically generates the service name from the class name
    if not provided.

    Args:
        obj: The class to decorate.

    Returns:
        The decorated class registered as a Pod.
    """
    if self.service_name is None:
        self.service_name = obj.__name__
    return super().__call__(obj)

데코레이터

RPC method decorator for gRPC service methods.

Provides the @rpc decorator for marking controller methods as gRPC service methods with support for all four gRPC streaming patterns.

RpcMethodType

Bases: StrEnum

gRPC method streaming patterns.

Attributes:

Name	Type	Description
UNARY		Single request, single response.
SERVER_STREAMING		Single request, stream of responses.
CLIENT_STREAMING		Stream of requests, single response.
BIDI_STREAMING		Stream of requests, stream of responses.

Rpc(method_type=RpcMethodType.UNARY, request_type=None, response_type=None) dataclass

Bases: FunctionAnnotation

Function annotation for marking methods as gRPC RPC endpoints.

Stores RPC configuration including the streaming pattern and request/response type metadata.

Attributes:

Name	Type	Description
method_type	RpcMethodType	gRPC streaming pattern for this method.
request_type	type[BaseModel] | None	Request message model, or None when the method declares none. Auto-extracted from type hints if not provided.
response_type	type[BaseModel] | None	Response message model, or None when the method declares none. Auto-extracted from type hints if not provided.

__call__(obj)

Annotate a method as an RPC endpoint.

Extracts request and response types from type hints if not explicitly provided.

Parameters:

Name	Type	Description	Default
obj	Callable[..., T]	The method to annotate.	required

Returns:

Type	Description
Callable[..., T]	The annotated method.

Source code in plugins/spakky-grpc/src/spakky/plugins/grpc/decorators/rpc.py

def __call__[T](self, obj: Callable[..., T]) -> Callable[..., T]:
    """Annotate a method as an RPC endpoint.

    Extracts request and response types from type hints if not
    explicitly provided.

    Args:
        obj: The method to annotate.

    Returns:
        The annotated method.
    """
    if self.request_type is None or self.response_type is None:
        self._extract_types(obj)
    return super().__call__(obj)

rpc(method_type=RpcMethodType.UNARY, request_type=None, response_type=None)

Decorator to mark a controller method as a gRPC RPC endpoint.

Attaches RPC configuration to the method including streaming pattern and message type metadata.

Parameters:

Name	Type	Description	Default
method_type	RpcMethodType	gRPC streaming pattern for this method.	UNARY
request_type	type[BaseModel] | None	Request message model. Auto-extracted from type hints if not provided.	None
response_type	type[BaseModel] | None	Response message model. Auto-extracted from type hints if not provided.	None

Returns:

Type	Description
Callable[[Callable[..., T]], Callable[..., T]]	A decorator function that attaches the RPC configuration.

Source code in plugins/spakky-grpc/src/spakky/plugins/grpc/decorators/rpc.py

def rpc[T](
    method_type: RpcMethodType = RpcMethodType.UNARY,
    request_type: type[BaseModel] | None = None,
    response_type: type[BaseModel] | None = None,
) -> Callable[[Callable[..., T]], Callable[..., T]]:
    """Decorator to mark a controller method as a gRPC RPC endpoint.

    Attaches RPC configuration to the method including streaming pattern
    and message type metadata.

    Args:
        method_type: gRPC streaming pattern for this method.
        request_type: Request message model. Auto-extracted from type hints
            if not provided.
        response_type: Response message model. Auto-extracted from type hints
            if not provided.

    Returns:
        A decorator function that attaches the RPC configuration.
    """

    def wrapper(method: Callable[..., T]) -> Callable[..., T]:
        return Rpc(
            method_type=method_type,
            request_type=request_type,
            response_type=response_type,
        )(method)

    return wrapper

어노테이션

Protobuf field annotation for pydantic-based message definitions.

Provides the ProtoField annotation for specifying protobuf field numbers on pydantic BaseModel fields using the Annotated type hint pattern.

Example::

from pydantic import BaseModel
from typing import Annotated

class HelloRequest(BaseModel):
    name: Annotated[str, ProtoField(number=1)]
    greeting_count: Annotated[int, ProtoField(number=2)]

ProtoField(number) dataclass

Protobuf field number annotation for pydantic model fields.

Used with typing.Annotated to specify the protobuf field number for a pydantic BaseModel field, enabling code-first protobuf message definition. The annotation is read at runtime from the model's model_fields[name].metadata tuple.

Attributes:

Name	Type	Description
number	int	The protobuf field number. Must be a positive integer.

number instance-attribute

The protobuf field number.

Handler

Generic RPC handler for code-first gRPC service dispatch.

Routes incoming gRPC calls to @GrpcController methods by matching the fully-qualified method name, performing protobuf ↔ pydantic BaseModel conversion via the google.protobuf.json_format bridge.

GrpcServiceHandler(*, controller_type, package, service_name, container, application_context, registry)

Bases: GenericRpcHandler

Generic handler dispatching gRPC calls to @GrpcController methods.

For each @rpc-decorated method, builds a grpc.RpcMethodHandler with serialiser/deserialiser that convert between protobuf wire format and pydantic BaseModel instances.

Attributes:

Name	Type	Description
_full_service_name	str	Fully-qualified <package>.<service> name.
_controller_type	type	The @GrpcController class.
_container	IContainer	IoC container for obtaining fresh controller instances.
_application_context	IApplicationContext	Application context for request-scoped isolation.
_registry	DescriptorRegistry	Descriptor registry for message class lookup.
_handlers	dict[str, RpcMethodHandler]	Pre-built map of /<package>.<service>/<method> → RpcMethodHandler.

Initialise the handler and pre-build per-method dispatchers.

Parameters:

Name	Type	Description	Default
controller_type	type	The @GrpcController-decorated class.	required
package	str	Protobuf package name.	required
service_name	str	gRPC service name.	required
container	IContainer	IoC container for obtaining controller instances.	required
application_context	IApplicationContext	Application context for request isolation.	required
registry	DescriptorRegistry	Descriptor registry for message class lookup.	required

Source code in plugins/spakky-grpc/src/spakky/plugins/grpc/handler.py

def __init__(
    self,
    *,
    controller_type: type,
    package: str,
    service_name: str,
    container: IContainer,
    application_context: IApplicationContext,
    registry: DescriptorRegistry,
) -> None:
    """Initialise the handler and pre-build per-method dispatchers.

    Args:
        controller_type: The ``@GrpcController``-decorated class.
        package: Protobuf package name.
        service_name: gRPC service name.
        container: IoC container for obtaining controller instances.
        application_context: Application context for request isolation.
        registry: Descriptor registry for message class lookup.
    """
    self._full_service_name = f"{package}.{service_name}"
    self._controller_type = controller_type
    self._container = container
    self._application_context = application_context
    self._registry = registry
    self._handlers = {}
    self._build_handlers()

service(handler_call_details)

Resolve an RPC method handler for the incoming call.

Parameters:

Name	Type	Description	Default
handler_call_details	HandlerCallDetails	Describes the incoming RPC.	required

Returns:

Type	Description
RpcMethodHandler | None	The matched handler, or None if not handled.

Source code in plugins/spakky-grpc/src/spakky/plugins/grpc/handler.py

@override
def service(
    self,
    handler_call_details: grpc.HandlerCallDetails,
) -> grpc.RpcMethodHandler | None:
    """Resolve an RPC method handler for the incoming call.

    Args:
        handler_call_details: Describes the incoming RPC.

    Returns:
        The matched handler, or ``None`` if not handled.
    """
    return self._handlers.get(handler_call_details.method)

클라이언트

Typed caller for a code-first @GrpcController service.

Without this, a caller has to rebuild the protobuf descriptor from its own copy of the pydantic models. Because field numbers are derived from field names, a single renamed field on either side moves that field's wire number and the value silently disappears from the decoded message.

:class:GrpcClient removes the second copy: it takes the controller class itself — the same declaration the server registers — builds the descriptor from it, and derives every callable's serialiser from that descriptor. The RPC method is identified by referencing the controller method rather than by spelling its name in a string, so a renamed method breaks at import time instead of at call time.

GrpcClient(channel, controller_type, registry=None)

Builds gRPC callables for one @GrpcController service.

The multicallable return types are written as strings because the gRPC runtime classes are only generic in the type stubs — subscripting them at class-definition time raises TypeError.

Attributes:

Name	Type	Description
registry	DescriptorRegistry	Registry holding the descriptor built from the controller.

Register the controller's descriptor and bind the client to a channel.

Parameters:

Name	Type	Description	Default
channel	Channel	Open channel to the server hosting the service.	required
controller_type	type	The @GrpcController-decorated class declaring the service.	required
registry	DescriptorRegistry | None	Registry to compile the descriptor into. None creates a private one; pass the server's registry when calling a service that runs in the same process.	None

Source code in plugins/spakky-grpc/src/spakky/plugins/grpc/client.py

def __init__(
    self,
    channel: grpc.aio.Channel,
    controller_type: type,
    registry: DescriptorRegistry | None = None,
) -> None:
    """Register the controller's descriptor and bind the client to a channel.

    Args:
        channel: Open channel to the server hosting the service.
        controller_type: The ``@GrpcController``-decorated class declaring
            the service.
        registry: Registry to compile the descriptor into. ``None`` creates
            a private one; pass the server's registry when calling a
            service that runs in the same process.
    """
    annotation = GrpcController.get(controller_type)
    self._channel = channel
    self._package = annotation.package
    self._full_service_name = (
        f"{annotation.package}."
        f"{annotation.service_name or controller_type.__name__}"
    )
    self.registry = registry if registry is not None else DescriptorRegistry()

    file_descriptor = build_file_descriptor(controller_type)
    if not self.registry.is_registered(file_descriptor.name):
        self.registry.register(file_descriptor)

unary_unary(method)

Build a callable for a single-request, single-response method.

Parameters:

Name	Type	Description	Default
method	Callable[[SelfT, RequestT], Coroutine[object, object, ResponseT]]	The @rpc method on the controller class, referenced unbound (EchoController.unary_echo).	required

Returns:

Type	Description
UnaryUnaryMultiCallable[RequestT, ResponseT]	A multicallable accepting the request model and awaiting the
UnaryUnaryMultiCallable[RequestT, ResponseT]	response model.

Source code in plugins/spakky-grpc/src/spakky/plugins/grpc/client.py

def unary_unary[SelfT, RequestT: BaseModel, ResponseT: BaseModel](
    self,
    method: Callable[[SelfT, RequestT], Coroutine[object, object, ResponseT]],
) -> "grpc.aio.UnaryUnaryMultiCallable[RequestT, ResponseT]":
    """Build a callable for a single-request, single-response method.

    Args:
        method: The ``@rpc`` method on the controller class, referenced
            unbound (``EchoController.unary_echo``).

    Returns:
        A multicallable accepting the request model and awaiting the
        response model.
    """
    method_path, serialize, deserialize = self._multicallable_arguments(
        method, RpcMethodType.UNARY
    )
    return self._channel.unary_unary(
        method_path,
        request_serializer=serialize,
        response_deserializer=deserialize,
    )

unary_stream(method)

Build a callable for a server-streaming method.

Parameters:

Name	Type	Description	Default
method	Callable[[SelfT, RequestT], AsyncIterator[ResponseT]]	The @rpc method on the controller class, referenced unbound.	required

Returns:

Type	Description
UnaryStreamMultiCallable[RequestT, ResponseT]	A multicallable accepting the request model and yielding response
UnaryStreamMultiCallable[RequestT, ResponseT]	models.

Source code in plugins/spakky-grpc/src/spakky/plugins/grpc/client.py

def unary_stream[SelfT, RequestT: BaseModel, ResponseT: BaseModel](
    self,
    method: Callable[[SelfT, RequestT], AsyncIterator[ResponseT]],
) -> "grpc.aio.UnaryStreamMultiCallable[RequestT, ResponseT]":
    """Build a callable for a server-streaming method.

    Args:
        method: The ``@rpc`` method on the controller class, referenced
            unbound.

    Returns:
        A multicallable accepting the request model and yielding response
        models.
    """
    method_path, serialize, deserialize = self._multicallable_arguments(
        method, RpcMethodType.SERVER_STREAMING
    )
    return self._channel.unary_stream(
        method_path,
        request_serializer=serialize,
        response_deserializer=deserialize,
    )

stream_unary(method)

Build a callable for a client-streaming method.

Parameters:

Name	Type	Description	Default
method	Callable[[SelfT, AsyncIterator[RequestT]], Coroutine[object, object, ResponseT]]	The @rpc method on the controller class, referenced unbound.	required

Returns:

Type	Description
StreamUnaryMultiCallable[RequestT, ResponseT]	A multicallable accepting an async iterator of request models and
StreamUnaryMultiCallable[RequestT, ResponseT]	awaiting the response model.

Source code in plugins/spakky-grpc/src/spakky/plugins/grpc/client.py

def stream_unary[SelfT, RequestT: BaseModel, ResponseT: BaseModel](
    self,
    method: Callable[
        [SelfT, AsyncIterator[RequestT]], Coroutine[object, object, ResponseT]
    ],
) -> "grpc.aio.StreamUnaryMultiCallable[RequestT, ResponseT]":
    """Build a callable for a client-streaming method.

    Args:
        method: The ``@rpc`` method on the controller class, referenced
            unbound.

    Returns:
        A multicallable accepting an async iterator of request models and
        awaiting the response model.
    """
    method_path, serialize, deserialize = self._multicallable_arguments(
        method, RpcMethodType.CLIENT_STREAMING
    )
    return self._channel.stream_unary(
        method_path,
        request_serializer=serialize,
        response_deserializer=deserialize,
    )

stream_stream(method)

Build a callable for a bidirectional-streaming method.

Parameters:

Name	Type	Description	Default
method	Callable[[SelfT, AsyncIterator[RequestT]], AsyncIterator[ResponseT]]	The @rpc method on the controller class, referenced unbound.	required

Returns:

Type	Description
StreamStreamMultiCallable[RequestT, ResponseT]	A multicallable accepting an async iterator of request models and
StreamStreamMultiCallable[RequestT, ResponseT]	yielding response models.

Source code in plugins/spakky-grpc/src/spakky/plugins/grpc/client.py

def stream_stream[SelfT, RequestT: BaseModel, ResponseT: BaseModel](
    self,
    method: Callable[[SelfT, AsyncIterator[RequestT]], AsyncIterator[ResponseT]],
) -> "grpc.aio.StreamStreamMultiCallable[RequestT, ResponseT]":
    """Build a callable for a bidirectional-streaming method.

    Args:
        method: The ``@rpc`` method on the controller class, referenced
            unbound.

    Returns:
        A multicallable accepting an async iterator of request models and
        yielding response models.
    """
    method_path, serialize, deserialize = self._multicallable_arguments(
        method, RpcMethodType.BIDI_STREAMING
    )
    return self._channel.stream_stream(
        method_path,
        request_serializer=serialize,
        response_deserializer=deserialize,
    )

메시지 변환

protobuf ↔ pydantic BaseModel translation shared by server and client.

Both the server-side generic handler and the client-side helper must encode the wire payload identically: a mismatch between the two directions silently drops or corrupts fields. Keeping the translation in one module makes that symmetry structural rather than a convention two call sites have to remember.

Every conversion routes through the google.protobuf.json_format bridge, using the message classes compiled from the shared :class:~spakky.plugins.grpc.schema.registry.DescriptorRegistry so that field numbers always come from the same descriptor the peer uses.

basemodel_to_protobuf(model, message_class)

Convert a pydantic BaseModel instance into a protobuf Message.

The model is serialised to JSON via pydantic's v2 model_dump_json API and parsed into a protobuf Message by json_format.Parse. None values from optional fields are emitted as JSON null which json_format treats as "field unset" for proto3 optional fields.

Parameters:

Name	Type	Description	Default
model	BaseModel	The pydantic BaseModel instance to convert.	required
message_class	type[Message]	The target protobuf message class.	required

Returns:

Type	Description
Message	A populated protobuf Message.

Source code in plugins/spakky-grpc/src/spakky/plugins/grpc/codec.py

def basemodel_to_protobuf(model: BaseModel, message_class: type[Message]) -> Message:
    """Convert a pydantic ``BaseModel`` instance into a protobuf ``Message``.

    The model is serialised to JSON via pydantic's v2 ``model_dump_json`` API
    and parsed into a protobuf ``Message`` by ``json_format.Parse``. ``None``
    values from optional fields are emitted as JSON ``null`` which
    ``json_format`` treats as "field unset" for proto3 optional fields.

    Args:
        model: The pydantic ``BaseModel`` instance to convert.
        message_class: The target protobuf message class.

    Returns:
        A populated protobuf ``Message``.
    """
    return json_format.Parse(
        model.model_dump_json(),
        message_class(),
        ignore_unknown_fields=False,
    )

protobuf_to_basemodel(message, model_type)

Convert a protobuf Message into a pydantic BaseModel instance.

The message is serialised to JSON with preserving_proto_field_name=True so field names round-trip unchanged into model_validate_json.

Parameters:

Name	Type	Description	Default
message	Message	The protobuf message.	required
model_type	type[BaseModelT]	The target BaseModel subclass.	required

Returns:

Type	Description
BaseModelT	An instance of model_type populated from message.

Source code in plugins/spakky-grpc/src/spakky/plugins/grpc/codec.py

def protobuf_to_basemodel[BaseModelT: BaseModel](
    message: Message, model_type: type[BaseModelT]
) -> BaseModelT:
    """Convert a protobuf ``Message`` into a pydantic ``BaseModel`` instance.

    The message is serialised to JSON with ``preserving_proto_field_name=True``
    so field names round-trip unchanged into ``model_validate_json``.

    Args:
        message: The protobuf message.
        model_type: The target ``BaseModel`` subclass.

    Returns:
        An instance of ``model_type`` populated from ``message``.
    """
    payload = json_format.MessageToJson(
        message,
        preserving_proto_field_name=True,
        always_print_fields_with_no_presence=True,
    )
    return model_type.model_validate_json(payload)

serializer_for(registry, full_name)

Return a BaseModel → wire-bytes serialiser for a registered type.

Parameters:

Name	Type	Description	Default
registry	DescriptorRegistry	Registry holding the compiled descriptor for full_name.	required
full_name	str	Fully-qualified protobuf message name.	required

Returns:

Type	Description
Callable[[BaseModel], bytes]	A callable encoding a BaseModel into protobuf wire bytes.

Source code in plugins/spakky-grpc/src/spakky/plugins/grpc/codec.py

def serializer_for(
    registry: DescriptorRegistry, full_name: str
) -> Callable[[BaseModel], bytes]:
    """Return a ``BaseModel`` → wire-bytes serialiser for a registered type.

    Args:
        registry: Registry holding the compiled descriptor for *full_name*.
        full_name: Fully-qualified protobuf message name.

    Returns:
        A callable encoding a ``BaseModel`` into protobuf wire bytes.
    """
    message_class = registry.get_message_class(full_name)

    def _serialize(model: BaseModel) -> bytes:
        return basemodel_to_protobuf(model, message_class).SerializeToString()

    return _serialize

deserializer_for(registry, full_name, model_type)

Return a wire-bytes → BaseModel deserialiser for a registered type.

Parameters:

Name	Type	Description	Default
registry	DescriptorRegistry	Registry holding the compiled descriptor for full_name.	required
full_name	str	Fully-qualified protobuf message name.	required
model_type	type[BaseModelT]	The BaseModel subclass to decode into.	required

Returns:

Type	Description
Callable[[bytes], BaseModelT]	A callable decoding protobuf wire bytes into model_type.

Source code in plugins/spakky-grpc/src/spakky/plugins/grpc/codec.py

def deserializer_for[BaseModelT: BaseModel](
    registry: DescriptorRegistry,
    full_name: str,
    model_type: type[BaseModelT],
) -> Callable[[bytes], BaseModelT]:
    """Return a wire-bytes → ``BaseModel`` deserialiser for a registered type.

    Args:
        registry: Registry holding the compiled descriptor for *full_name*.
        full_name: Fully-qualified protobuf message name.
        model_type: The ``BaseModel`` subclass to decode into.

    Returns:
        A callable decoding protobuf wire bytes into *model_type*.
    """
    message_class = registry.get_message_class(full_name)

    def _deserialize(data: bytes) -> BaseModelT:
        message = message_class()
        message.ParseFromString(data)
        return protobuf_to_basemodel(message, model_type)

    return _deserialize

인증 경계

gRPC auth boundary helpers.

seed_grpc_auth_context(*, container, application_context, context, operation)

Authenticate gRPC metadata and seed AuthContext when credentials exist.

Source code in plugins/spakky-grpc/src/spakky/plugins/grpc/auth.py

def seed_grpc_auth_context(
    *,
    container: IContainer,
    application_context: IApplicationContext,
    context: grpc.aio.ServicerContext,
    operation: str,
) -> None:
    """Authenticate gRPC metadata and seed AuthContext when credentials exist."""
    credential = _extract_credential(context.invocation_metadata())
    if credential is None:
        return
    provider = container.get_or_none(IAuthenticationProvider)
    if provider is None:
        raise Unavailable()
    auth_context = provider.authenticate(
        credential,
        AuthInvocation(boundary=GRPC_AUTH_BOUNDARY, operation=operation),
    )
    store_auth_context(application_context, auth_context)

설정

gRPC plugin configuration.

SPAKKY_GRPC_CONFIG_ENV_PREFIX = 'SPAKKY_GRPC_' module-attribute

Environment prefix for gRPC plugin settings.

GrpcServerOptions = dict[str, int | str]

gRPC channel arguments keyed by their documented grpc.* option name.

GrpcConfig()

Bases: BaseSettings

Configuration for the gRPC server integration.

Source code in plugins/spakky-grpc/src/spakky/plugins/grpc/config.py

def __init__(self) -> None:
    super().__init__()

bind_addresses = () class-attribute instance-attribute

Addresses the server listens on, each in host:port form.

server_options = {} class-attribute instance-attribute

Channel arguments forwarded verbatim to grpc.aio.server(options=...).

Any documented gRPC channel argument is accepted, so keepalive intervals (grpc.keepalive_time_ms), message size caps (grpc.max_receive_message_length) and connection lifetime limits (grpc.max_connection_age_ms) are all tunable without this plugin enumerating them one by one.

tls_certificate_chain_file = None class-attribute instance-attribute

PEM file with the server certificate chain. None means TLS is off.

tls_private_key_file = None class-attribute instance-attribute

PEM file with the private key matching the certificate chain.

tls_client_ca_file = None class-attribute instance-attribute

PEM file of authorities trusted to sign client certificates (mutual TLS).

require_client_auth = False class-attribute instance-attribute

Whether clients must present a certificate signed by tls_client_ca_file.

health_service_enabled = True class-attribute instance-attribute

Whether to expose the standard grpc.health.v1.Health service.

reflection_service_enabled = True class-attribute instance-attribute

Whether to expose the standard server reflection service.

전송 보안

Transport security credentials for the gRPC listener.

Translates the PEM file paths declared in :class:GrpcConfig into a grpc.ServerCredentials object. A half-configured key pair is rejected up front rather than silently degrading to a plaintext listener, because a deployment that asked for TLS must never end up serving cleartext.

build_server_credentials(config)

Build listener credentials from the configured PEM files.

Parameters:

Name	Type	Description	Default
config	GrpcConfig	Plugin configuration carrying the TLS file paths.	required

Returns:

Type	Description
ServerCredentials | None	Credentials for a TLS listener, or None when no transport
ServerCredentials | None	security setting is present and the listener stays plaintext.

Raises:

Type	Description
IncompleteTlsCredentialsError	If some transport security setting is present but the certificate chain or private key is missing.
MissingClientCertificateAuthorityError	If client certificate authentication is required without a client CA file.

Source code in plugins/spakky-grpc/src/spakky/plugins/grpc/credentials.py

def build_server_credentials(config: GrpcConfig) -> grpc.ServerCredentials | None:
    """Build listener credentials from the configured PEM files.

    Args:
        config: Plugin configuration carrying the TLS file paths.

    Returns:
        Credentials for a TLS listener, or ``None`` when no transport
        security setting is present and the listener stays plaintext.

    Raises:
        IncompleteTlsCredentialsError: If some transport security setting is
            present but the certificate chain or private key is missing.
        MissingClientCertificateAuthorityError: If client certificate
            authentication is required without a client CA file.
    """
    if not _has_transport_security_setting(config):
        return None
    if config.tls_certificate_chain_file is None or config.tls_private_key_file is None:
        raise IncompleteTlsCredentialsError(
            config.tls_certificate_chain_file,
            config.tls_private_key_file,
        )
    if config.require_client_auth and config.tls_client_ca_file is None:
        raise MissingClientCertificateAuthorityError

    return grpc.ssl_server_credentials(
        [
            (
                config.tls_private_key_file.read_bytes(),
                config.tls_certificate_chain_file.read_bytes(),
            )
        ],
        root_certificates=(
            config.tls_client_ca_file.read_bytes()
            if config.tls_client_ca_file is not None
            else None
        ),
        require_client_auth=config.require_client_auth,
    )

서버 명세

Deferred gRPC server configuration.

grpc.aio.server() binds to the current event loop at creation time, so the real server must be instantiated on the event loop that eventually runs it. :class:GrpcServerSpec collects everything needed to build the server (interceptors, generic handlers, bind targets, channel arguments, standard-service registrations) during post-processing, and :class:GrpcServerService materialises it at start_async time on the correct loop.

ServerServiceRegistrar = Callable[[grpc.aio.Server], Awaitable[None]]

Callback attaching a service to the server once it has been instantiated.

The standard health and reflection services ship as servicers that must be registered on a concrete grpc.aio.Server, which only exists after :meth:GrpcServerSpec.build_async. Collecting them as callbacks keeps that registration on the same deferred timeline as everything else in the spec. They are awaitable because reporting an initial health status goes through the servicer's async API on the serving loop.

GrpcBindTarget(address, credentials) dataclass

One listener address together with the credentials protecting it.

grpc.ServerCredentials is an opaque handle from the gRPC C core, so this pairing is a plain dataclass rather than a pydantic model.

Attributes:

Name	Type	Description
address	str	Listener address in host:port form.
credentials	ServerCredentials | None	TLS credentials, or None for a plaintext listener.

GrpcServerSpec(options=None)

Configuration collected during post-processing for deferred server creation.

Attributes:

Name	Type	Description
handlers	list[GenericRpcHandler]	Generic RPC handlers to register on the server.
interceptors	list[ServerInterceptor]	Server interceptors to apply at creation time.
bind_targets	list[GrpcBindTarget]	Listener addresses with their transport credentials.
service_registrars	list[ServerServiceRegistrar]	Callbacks attaching standard services once the server object exists.
options	tuple[tuple[str, int | str], ...]	Channel arguments passed to grpc.aio.server.
bound_ports	list[int]	Ports returned when each bind target is attached, populated when :meth:build_async runs. Useful when binding to :0 and needing to discover the OS-assigned port.

Initialise an empty spec.

Parameters:

Name	Type	Description	Default
options	Mapping[str, int | str] | None	Channel arguments for grpc.aio.server. None keeps the gRPC defaults.	None

Source code in plugins/spakky-grpc/src/spakky/plugins/grpc/server_spec.py

def __init__(self, options: Mapping[str, int | str] | None = None) -> None:
    """Initialise an empty spec.

    Args:
        options: Channel arguments for ``grpc.aio.server``. ``None`` keeps
            the gRPC defaults.
    """
    self.handlers = []
    self.interceptors = []
    self.bind_targets = []
    self.service_registrars = []
    self.options = tuple(options.items()) if options is not None else ()
    self.bound_ports = []

bind_addresses property

Listener addresses in registration order.

add_handler(handler)

Register a generic RPC handler.

Parameters:

Name	Type	Description	Default
handler	GenericRpcHandler	The handler to add to the server.	required

Source code in plugins/spakky-grpc/src/spakky/plugins/grpc/server_spec.py

def add_handler(self, handler: grpc.GenericRpcHandler) -> None:
    """Register a generic RPC handler.

    Args:
        handler: The handler to add to the server.
    """
    self.handlers.append(handler)

add_interceptor(interceptor)

Register a server interceptor.

Parameters:

Name	Type	Description	Default
interceptor	ServerInterceptor	The interceptor to install on the server.	required

Source code in plugins/spakky-grpc/src/spakky/plugins/grpc/server_spec.py

def add_interceptor(self, interceptor: grpc.aio.ServerInterceptor) -> None:
    """Register a server interceptor.

    Args:
        interceptor: The interceptor to install on the server.
    """
    self.interceptors.append(interceptor)

add_service_registrar(registrar)

Register a callback that attaches a service at build time.

Parameters:

Name	Type	Description	Default
registrar	ServerServiceRegistrar	Callback invoked with the instantiated server.	required

Source code in plugins/spakky-grpc/src/spakky/plugins/grpc/server_spec.py

def add_service_registrar(self, registrar: ServerServiceRegistrar) -> None:
    """Register a callback that attaches a service at build time.

    Args:
        registrar: Callback invoked with the instantiated server.
    """
    self.service_registrars.append(registrar)

add_insecure_port(address)

Register a plaintext bind address.

Parameters:

Name	Type	Description	Default
address	str	Address in host:port form.	required

Source code in plugins/spakky-grpc/src/spakky/plugins/grpc/server_spec.py

def add_insecure_port(self, address: str) -> None:
    """Register a plaintext bind address.

    Args:
        address: Address in ``host:port`` form.
    """
    self.bind_targets.append(GrpcBindTarget(address=address, credentials=None))

add_secure_port(address, credentials)

Register a TLS-protected bind address.

Parameters:

Name	Type	Description	Default
address	str	Address in host:port form.	required
credentials	ServerCredentials	Credentials terminating TLS on that address.	required

Source code in plugins/spakky-grpc/src/spakky/plugins/grpc/server_spec.py

def add_secure_port(
    self, address: str, credentials: grpc.ServerCredentials
) -> None:
    """Register a TLS-protected bind address.

    Args:
        address: Address in ``host:port`` form.
        credentials: Credentials terminating TLS on that address.
    """
    self.bind_targets.append(
        GrpcBindTarget(address=address, credentials=credentials)
    )

build_async() async

Instantiate the underlying grpc.aio.Server on the current loop.

Must be awaited from the event loop that will run the server; see the module docstring for the rationale.

Returns:

Type	Description
Server	The fully-configured server ready for .start().

Source code in plugins/spakky-grpc/src/spakky/plugins/grpc/server_spec.py

async def build_async(self) -> grpc.aio.Server:
    """Instantiate the underlying ``grpc.aio.Server`` on the current loop.

    Must be awaited from the event loop that will run the server; see
    the module docstring for the rationale.

    Returns:
        The fully-configured server ready for ``.start()``.
    """
    server = grpc.aio.server(
        interceptors=list(self.interceptors),
        options=self.options,
    )
    server.add_generic_rpc_handlers(tuple(self.handlers))
    for register_service in self.service_registrars:
        await register_service(server)
    self.bound_ports = [self._bind(server, target) for target in self.bind_targets]
    return server

표준 서비스

Standard gRPC services exposed alongside the application's own services.

Two services come from the gRPC ecosystem rather than from user code:

• grpc.health.v1.Health is what Kubernetes' native gRPC probe calls, so a gRPC-only deployment has no other way to report liveness.
• Server reflection is the only way to discover what a running server exposes when the schema is built at runtime from pydantic models and no .proto artifact is ever produced.

Both are attached through :meth:GrpcServerSpec.add_service_registrar, because their servicers need the concrete grpc.aio.Server that only exists once the spec is built on the serving event loop.

enable_health_service(spec, registry, servicer)

Expose the standard health checking service on the server.

Every registered service is reported SERVING when the server comes up. A probe that names a service — grpc_health_probe -service=<name> or the Kubernetes grpc.service field — answers NOT_FOUND for any name the servicer has never been told about, so a healthy server would otherwise be judged unready. Applications flip individual services afterwards through the injected servicer.

Parameters:

Name	Type	Description	Default
spec	GrpcServerSpec	Spec collecting the deferred server configuration.	required
registry	DescriptorRegistry	Registry the health schema is mirrored into so reflection can describe the service it advertises, and whose service list seeds the initial statuses.	required
servicer	HealthServicer	Servicer holding the reported serving statuses.	required

Source code in plugins/spakky-grpc/src/spakky/plugins/grpc/standard_services.py

def enable_health_service(
    spec: GrpcServerSpec,
    registry: DescriptorRegistry,
    servicer: health.aio.HealthServicer,
) -> None:
    """Expose the standard health checking service on the server.

    Every registered service is reported ``SERVING`` when the server comes up.
    A probe that names a service — ``grpc_health_probe -service=<name>`` or the
    Kubernetes ``grpc.service`` field — answers ``NOT_FOUND`` for any name the
    servicer has never been told about, so a healthy server would otherwise be
    judged unready. Applications flip individual services afterwards through
    the injected servicer.

    Args:
        spec: Spec collecting the deferred server configuration.
        registry: Registry the health schema is mirrored into so reflection
            can describe the service it advertises, and whose service list
            seeds the initial statuses.
        servicer: Servicer holding the reported serving statuses.
    """
    _mirror_descriptor(registry, health_pb2.DESCRIPTOR)

    async def _register(server: grpc.aio.Server) -> None:
        health_pb2_grpc.add_HealthServicer_to_server(servicer, server)
        for service_name in registry.service_names:
            await servicer.set(service_name, health_pb2.HealthCheckResponse.SERVING)

    spec.add_service_registrar(_register)

enable_reflection_service(spec, registry)

Expose server reflection over the code-first descriptor pool.

Parameters:

Name	Type	Description	Default
spec	GrpcServerSpec	Spec collecting the deferred server configuration.	required
registry	DescriptorRegistry	Registry holding every descriptor built from controllers; its service list is read at build time so services registered later during post-processing are still advertised.	required

Source code in plugins/spakky-grpc/src/spakky/plugins/grpc/standard_services.py

def enable_reflection_service(
    spec: GrpcServerSpec, registry: DescriptorRegistry
) -> None:
    """Expose server reflection over the code-first descriptor pool.

    Args:
        spec: Spec collecting the deferred server configuration.
        registry: Registry holding every descriptor built from controllers;
            its service list is read at build time so services registered
            later during post-processing are still advertised.
    """
    _mirror_descriptor(registry, reflection_pb2.DESCRIPTOR)

    async def _register(server: grpc.aio.Server) -> None:
        reflection.enable_server_reflection(
            registry.service_names, server, pool=registry.pool
        )

    spec.add_service_registrar(_register)

스키마

Registry

Descriptor pool registry for compiled protobuf descriptors.

Manages FileDescriptorProto registration in a descriptor_pool, provides caching, and returns compiled message classes and service descriptors.

DescriptorRegistry(pool=None)

Registry for protobuf descriptors backed by a DescriptorPool.

Registers FileDescriptorProto instances, prevents duplicates, and provides access to compiled message classes and service descriptors.

Attributes:

Name	Type	Description
pool	DescriptorPool	The underlying DescriptorPool.

Source code in plugins/spakky-grpc/src/spakky/plugins/grpc/schema/registry.py

def __init__(self, pool: DescriptorPool | None = None) -> None:
    self.pool: DescriptorPool = pool or DescriptorPool()
    self._registered_files: set[str] = set()
    self._service_names: set[str] = set()

service_names property

Fully-qualified names of every service registered so far, sorted.

Server reflection advertises exactly this list, so it is read at server-build time rather than captured earlier: controllers keep registering while post-processing runs.

register(file_proto)

Register a FileDescriptorProto in the pool.

Parameters:

Name	Type	Description	Default
file_proto	FileDescriptorProto	The file descriptor proto to register.	required

Returns:

Type	Description
FileDescriptor	The compiled FileDescriptor.

Raises:

Type	Description
DescriptorAlreadyRegisteredError	If the file is already registered.

Source code in plugins/spakky-grpc/src/spakky/plugins/grpc/schema/registry.py

def register(self, file_proto: FileDescriptorProto) -> FileDescriptor:
    """Register a FileDescriptorProto in the pool.

    Args:
        file_proto: The file descriptor proto to register.

    Returns:
        The compiled FileDescriptor.

    Raises:
        DescriptorAlreadyRegisteredError: If the file is already
            registered.
    """
    if file_proto.name in self._registered_files:
        raise DescriptorAlreadyRegisteredError(file_proto.name)

    self._registered_files.add(file_proto.name)
    self._service_names.update(
        f"{file_proto.package}.{service.name}" for service in file_proto.service
    )
    serialized = file_proto.SerializeToString()
    self.pool.AddSerializedFile(serialized)
    return self.pool.FindFileByName(file_proto.name)

is_registered(file_name)

Check if a file is already registered.

Parameters:

Name	Type	Description	Default
file_name	str	The proto file name.	required

Returns:

Type	Description
bool	True if the file has been registered.

Source code in plugins/spakky-grpc/src/spakky/plugins/grpc/schema/registry.py

def is_registered(self, file_name: str) -> bool:
    """Check if a file is already registered.

    Args:
        file_name: The proto file name.

    Returns:
        True if the file has been registered.
    """
    return file_name in self._registered_files

find_message_descriptor(full_name)

Find a message descriptor by its fully-qualified name.

Parameters:

Name	Type	Description	Default
full_name	str	The fully-qualified protobuf message name (e.g. package.MessageName).	required

Returns:

Type	Description
Descriptor	The message Descriptor.

Source code in plugins/spakky-grpc/src/spakky/plugins/grpc/schema/registry.py

def find_message_descriptor(self, full_name: str) -> Descriptor:
    """Find a message descriptor by its fully-qualified name.

    Args:
        full_name: The fully-qualified protobuf message name
            (e.g. ``package.MessageName``).

    Returns:
        The message Descriptor.
    """
    return self.pool.FindMessageTypeByName(full_name)

get_message_class(full_name)

Get a runtime message class for the given type name.

Parameters:

Name	Type	Description	Default
full_name	str	The fully-qualified protobuf message name.	required

Returns:

Type	Description
type[Message]	A Message subclass that can be instantiated.

Source code in plugins/spakky-grpc/src/spakky/plugins/grpc/schema/registry.py

def get_message_class(self, full_name: str) -> type[Message]:
    """Get a runtime message class for the given type name.

    Args:
        full_name: The fully-qualified protobuf message name.

    Returns:
        A Message subclass that can be instantiated.
    """
    descriptor = self.find_message_descriptor(full_name)
    return GetMessageClass(descriptor)

find_service_descriptor(full_name)

Find a service descriptor by its fully-qualified name.

Parameters:

Name	Type	Description	Default
full_name	str	The fully-qualified protobuf service name (e.g. package.ServiceName).	required

Returns:

Type	Description
ServiceDescriptor	The ServiceDescriptor.

Source code in plugins/spakky-grpc/src/spakky/plugins/grpc/schema/registry.py

def find_service_descriptor(self, full_name: str) -> ServiceDescriptor:
    """Find a service descriptor by its fully-qualified name.

    Args:
        full_name: The fully-qualified protobuf service name
            (e.g. ``package.ServiceName``).

    Returns:
        The ServiceDescriptor.
    """
    return self.pool.FindServiceByName(full_name)

Descriptor Builder

Pydantic BaseModel to protobuf FileDescriptorProto builder.

Converts pydantic BaseModel subclasses and @rpc-decorated controller methods into protobuf FileDescriptorProto instances. Field numbers are assigned deterministically from field names (see :mod:spakky.plugins.grpc.schema.field_number); an explicit ProtoField annotation overrides the derived number for a field.

build_message_descriptor(model_type, collected=None)

Build a DescriptorProto from a pydantic BaseModel subclass.

Recursively processes nested BaseModel fields into nested message descriptors.

Parameters:

Name	Type	Description	Default
model_type	type[BaseModel]	The BaseModel subclass to convert.	required
collected	dict[str, DescriptorProto] | None	Accumulator for all message descriptors encountered during recursive processing. Used internally.	None

Returns:

Type	Description
DescriptorProto	A tuple of (root DescriptorProto, dict of all collected
dict[str, DescriptorProto]	descriptors).

Source code in plugins/spakky-grpc/src/spakky/plugins/grpc/schema/descriptor_builder.py

def build_message_descriptor(
    model_type: type[BaseModel],
    collected: dict[str, DescriptorProto] | None = None,
) -> tuple[DescriptorProto, dict[str, DescriptorProto]]:
    """Build a ``DescriptorProto`` from a pydantic ``BaseModel`` subclass.

    Recursively processes nested ``BaseModel`` fields into nested message
    descriptors.

    Args:
        model_type: The ``BaseModel`` subclass to convert.
        collected: Accumulator for all message descriptors encountered
            during recursive processing. Used internally.

    Returns:
        A tuple of (root ``DescriptorProto``, dict of all collected
        descriptors).
    """
    if collected is None:
        collected = {}

    name = model_type.__name__
    if name in collected:
        return collected[name], collected

    descriptor = DescriptorProto(name=name)
    collected[name] = descriptor

    field_numbers = assign_field_numbers(model_type)

    for field_name, field_info in model_type.model_fields.items():
        resolved = resolve_type(field_info.annotation)

        field_desc = FieldDescriptorProto(
            name=field_name,
            number=field_numbers[field_name],
            type=resolved.proto_type,
        )

        if resolved.is_repeated:
            field_desc.label = FieldDescriptorProto.LABEL_REPEATED
        elif resolved.is_optional:
            field_desc.label = FieldDescriptorProto.LABEL_OPTIONAL
            field_desc.proto3_optional = True
            oneof_index = len(descriptor.oneof_decl)
            descriptor.oneof_decl.append(OneofDescriptorProto(name=f"__{field_name}"))
            field_desc.oneof_index = oneof_index
        else:
            field_desc.label = FieldDescriptorProto.LABEL_OPTIONAL

        if resolved.is_message and resolved.message_type is not None:
            field_desc.type_name = resolved.message_type.__name__
            build_message_descriptor(resolved.message_type, collected)

        descriptor.field.append(field_desc)

    return descriptor, collected

build_service_descriptor(controller_type, package, service_name, collected)

Build a ServiceDescriptorProto from an @GrpcController class.

Inspects all @rpc-decorated methods on the controller and generates method descriptors with fully-qualified type names.

Parameters:

Name	Type	Description	Default
controller_type	type	The controller class to inspect.	required
package	str	The protobuf package name.	required
service_name	str	The gRPC service name.	required
collected	dict[str, DescriptorProto]	Accumulator for message descriptors found in method signatures.	required

Returns:

Type	Description
ServiceDescriptorProto	A ServiceDescriptorProto for the controller.

Raises:

Type	Description
MessagelessRpcMethodError	If an @rpc method declares no request or no response model. protobuf requires every method to name an input and an output message, so such a declaration cannot be expressed as a descriptor at all.

Source code in plugins/spakky-grpc/src/spakky/plugins/grpc/schema/descriptor_builder.py

def build_service_descriptor(
    controller_type: type,
    package: str,
    service_name: str,
    collected: dict[str, DescriptorProto],
) -> ServiceDescriptorProto:
    """Build a ``ServiceDescriptorProto`` from an ``@GrpcController`` class.

    Inspects all ``@rpc``-decorated methods on the controller and generates
    method descriptors with fully-qualified type names.

    Args:
        controller_type: The controller class to inspect.
        package: The protobuf package name.
        service_name: The gRPC service name.
        collected: Accumulator for message descriptors found in method
            signatures.

    Returns:
        A ``ServiceDescriptorProto`` for the controller.

    Raises:
        MessagelessRpcMethodError: If an ``@rpc`` method declares no request
            or no response model. protobuf requires every method to name an
            input and an output message, so such a declaration cannot be
            expressed as a descriptor at all.
    """
    service = ServiceDescriptorProto(name=service_name)

    for method_name, method in getmembers(controller_type, predicate=isfunction):
        if not Rpc.exists(method):
            continue

        rpc_annotation = Rpc.get(method)
        request_type = rpc_annotation.request_type
        response_type = rpc_annotation.response_type
        if request_type is None or response_type is None:
            raise MessagelessRpcMethodError(f"{controller_type.__name__}.{method_name}")

        build_message_descriptor(request_type, collected)
        build_message_descriptor(response_type, collected)

        method_desc = MethodDescriptorProto(
            name=method_name,
            input_type=f".{package}.{request_type.__name__}",
            output_type=f".{package}.{response_type.__name__}",
            client_streaming=rpc_annotation.method_type
            in {RpcMethodType.CLIENT_STREAMING, RpcMethodType.BIDI_STREAMING},
            server_streaming=rpc_annotation.method_type
            in {RpcMethodType.SERVER_STREAMING, RpcMethodType.BIDI_STREAMING},
        )

        service.method.append(method_desc)

    return service

build_file_descriptor(controller_type)

Build a complete FileDescriptorProto from an @GrpcController class.

Generates all message descriptors referenced by @rpc methods and the service descriptor, packaged into a single FileDescriptorProto.

Parameters:

Name	Type	Description	Default
controller_type	type	The @GrpcController-decorated class.	required

Returns:

Type	Description
FileDescriptorProto	A FileDescriptorProto ready for descriptor_pool registration.

Raises:

Type	Description
MessagelessRpcMethodError	If an @rpc method declares no request or no response model.

Source code in plugins/spakky-grpc/src/spakky/plugins/grpc/schema/descriptor_builder.py

def build_file_descriptor(controller_type: type) -> FileDescriptorProto:
    """Build a complete ``FileDescriptorProto`` from an ``@GrpcController`` class.

    Generates all message descriptors referenced by ``@rpc`` methods and
    the service descriptor, packaged into a single ``FileDescriptorProto``.

    Args:
        controller_type: The ``@GrpcController``-decorated class.

    Returns:
        A ``FileDescriptorProto`` ready for ``descriptor_pool`` registration.

    Raises:
        MessagelessRpcMethodError: If an ``@rpc`` method declares no request
            or no response model.
    """
    annotation = GrpcController.get(controller_type)
    package = annotation.package
    service_name = annotation.service_name or controller_type.__name__

    file_name = f"{package.replace('.', '/')}/{service_name}.proto"

    collected: dict[str, DescriptorProto] = {}
    service = build_service_descriptor(
        controller_type, package, service_name, collected
    )

    file_desc = FileDescriptorProto(
        name=file_name,
        package=package,
        syntax="proto3",
    )

    for message_desc in collected.values():
        file_desc.message_type.append(message_desc)

    file_desc.service.append(service)

    return file_desc

타입 맵

Python type to protobuf type mapping.

Maps Python built-in types and composite types (list, Optional, nested BaseModel) to their protobuf FieldDescriptorProto equivalents. Protobuf field numbers are assigned separately by :mod:spakky.plugins.grpc.schema.field_number.

PYTHON_TO_PROTO_TYPE = {str: FieldDescriptorProto.TYPE_STRING, int: FieldDescriptorProto.TYPE_INT64, float: FieldDescriptorProto.TYPE_DOUBLE, bool: FieldDescriptorProto.TYPE_BOOL, bytes: FieldDescriptorProto.TYPE_BYTES} module-attribute

Mapping of Python primitive types to protobuf field type constants.

ResolvedFieldType(proto_type, *, is_repeated=False, is_optional=False, is_message=False, message_type=None)

Result of resolving a Python type annotation to protobuf metadata.

Attributes:

Name	Type	Description
proto_type		Protobuf field type constant from FieldDescriptorProto.
is_repeated		Whether the field is a repeated (list) field.
is_optional		Whether the field is optional.
is_message		Whether the field references a nested message type.
message_type		The nested BaseModel type for message fields.

Source code in plugins/spakky-grpc/src/spakky/plugins/grpc/schema/type_map.py

def __init__(
    self,
    proto_type: FieldDescriptorProto.Type.ValueType,
    *,
    is_repeated: bool = False,
    is_optional: bool = False,
    is_message: bool = False,
    message_type: type[BaseModel] | None = None,
) -> None:
    self.proto_type = proto_type
    self.is_repeated = is_repeated
    self.is_optional = is_optional
    self.is_message = is_message
    self.message_type = message_type

resolve_type(annotation)

Resolve a Python type annotation to protobuf field metadata.

Handles: - Primitive types (str, int, float, bool, bytes) - list[T] → repeated field - Optional[T] (T | None) → optional field - Nested BaseModel → message type

Parameters:

Name	Type	Description	Default
annotation	object	The Python type annotation to resolve.	required

Returns:

Type	Description
ResolvedFieldType	A ResolvedFieldType with protobuf mapping information.

Raises:

Type	Description
UnsupportedFieldTypeError	If the type cannot be mapped.

Source code in plugins/spakky-grpc/src/spakky/plugins/grpc/schema/type_map.py

def resolve_type(annotation: object) -> ResolvedFieldType:
    """Resolve a Python type annotation to protobuf field metadata.

    Handles:
    - Primitive types (str, int, float, bool, bytes)
    - ``list[T]`` → repeated field
    - ``Optional[T]`` (``T | None``) → optional field
    - Nested ``BaseModel`` → message type

    Args:
        annotation: The Python type annotation to resolve.

    Returns:
        A ResolvedFieldType with protobuf mapping information.

    Raises:
        UnsupportedFieldTypeError: If the type cannot be mapped.
    """
    origin = get_origin(annotation)
    args = get_args(annotation)

    if origin is list:
        inner = args[0] if args else None
        if inner is None:  # pragma: no cover - defensive guard, list[T] always has args
            raise UnsupportedFieldTypeError(list)
        inner_resolved = _resolve_scalar(inner)
        return ResolvedFieldType(
            proto_type=inner_resolved.proto_type,
            is_repeated=True,
            is_message=inner_resolved.is_message,
            message_type=inner_resolved.message_type,
        )

    if _is_union(annotation, origin, args):
        non_none = [a for a in args if a is not type(None)]
        if len(non_none) != 1:
            raise UnsupportedFieldTypeError(type(annotation))
        inner_resolved = _resolve_scalar(non_none[0])
        return ResolvedFieldType(
            proto_type=inner_resolved.proto_type,
            is_optional=True,
            is_message=inner_resolved.is_message,
            message_type=inner_resolved.message_type,
        )

    return _resolve_scalar(annotation)

Descriptor 스냅샷

Wire-layout snapshot of the descriptors generated from controllers.

Field numbers are derived from field names, so renaming a pydantic field is a wire-breaking change that no .proto artifact exists to catch. This module renders the generated layout — message → field → number and type — as deterministic JSON, and exposes it as the spakky-grpc-descriptor-snapshot command so a project can commit the snapshot and fail its own build when the wire layout moves.

ProtoFieldSnapshot

Bases: BaseModel

One protobuf field's wire-visible identity.

type_name instance-attribute

Fully-qualified type for message fields, empty for scalar fields.

proto3_optional instance-attribute

Whether the field tracks explicit presence.

Turning str into str | None keeps the number, the type and the label identical while changing what a peer decodes for an unset field between "" and None. Without this the snapshot diff would stay clean through that break.

ProtoMessageSnapshot

Bases: BaseModel

One protobuf message with its fields ordered by field number.

ProtoMethodSnapshot

Bases: BaseModel

One RPC method with its request/response types and streaming shape.

ProtoServiceSnapshot

Bases: BaseModel

One gRPC service with its methods ordered by name.

DescriptorSnapshot

Bases: BaseModel

The full wire layout generated from a set of controllers.

build_descriptor_snapshot(controller_types)

Render the wire layout generated from @GrpcController classes.

Parameters:

Name	Type	Description	Default
controller_types	Sequence[type]	Controller classes to describe.	required

Returns:

Type	Description
DescriptorSnapshot	A snapshot whose services, messages and fields are in a stable order
DescriptorSnapshot	so two runs over the same declarations produce identical output.

Source code in plugins/spakky-grpc/src/spakky/plugins/grpc/schema/snapshot.py

def build_descriptor_snapshot(
    controller_types: Sequence[type],
) -> DescriptorSnapshot:
    """Render the wire layout generated from ``@GrpcController`` classes.

    Args:
        controller_types: Controller classes to describe.

    Returns:
        A snapshot whose services, messages and fields are in a stable order
        so two runs over the same declarations produce identical output.
    """
    file_descriptors = [
        build_file_descriptor(controller_type) for controller_type in controller_types
    ]
    # Controllers sharing a message type each carry their own copy of it, so the
    # messages are keyed by qualified name to emit one entry per wire type.
    messages = {
        f"{file_descriptor.package}.{message.name}": _message_snapshot(
            file_descriptor.package, message
        )
        for file_descriptor in file_descriptors
        for message in file_descriptor.message_type
    }
    return DescriptorSnapshot(
        services=sorted(
            (
                _service_snapshot(file_descriptor.package, service)
                for file_descriptor in file_descriptors
                for service in file_descriptor.service
            ),
            key=lambda service: service.name,
        ),
        messages=sorted(messages.values(), key=lambda message: message.name),
    )

collect_controller_types(module_names)

Import the named modules and collect every @GrpcController in them.

Parameters:

Name	Type	Description	Default
module_names	Sequence[str]	Dotted module or package paths to scan.	required

Returns:

Type	Description
list[type]	The discovered controller classes, ordered by qualified name.

Raises:

Type	Description
NoControllerFoundError	If the named modules declare no controller. An empty snapshot would otherwise be committed as the baseline and every later comparison against it would pass, leaving the wire layout unguarded.

Source code in plugins/spakky-grpc/src/spakky/plugins/grpc/schema/snapshot.py

def collect_controller_types(module_names: Sequence[str]) -> list[type]:
    """Import the named modules and collect every ``@GrpcController`` in them.

    Args:
        module_names: Dotted module or package paths to scan.

    Returns:
        The discovered controller classes, ordered by qualified name.

    Raises:
        NoControllerFoundError: If the named modules declare no controller. An
            empty snapshot would otherwise be committed as the baseline and
            every later comparison against it would pass, leaving the wire
            layout unguarded.
    """
    controller_types: set[type] = set()
    for module_name in module_names:
        for module in _expand_module(resolve_module(module_name)):
            controller_types |= list_classes(module, GrpcController.exists)
    if not controller_types:
        raise NoControllerFoundError(tuple(module_names))
    return sorted(controller_types, key=lambda controller: controller.__qualname__)

main(argv=None)

Print the descriptor snapshot for the modules named on the command line.

Parameters:

Name	Type	Description	Default
argv	Sequence[str] | None	Command-line arguments. None reads sys.argv.	None

Returns:

Type	Description
int	The process exit status, which is always success once the modules
int	imported and the descriptors built.

Source code in plugins/spakky-grpc/src/spakky/plugins/grpc/schema/snapshot.py

def main(argv: Sequence[str] | None = None) -> int:
    """Print the descriptor snapshot for the modules named on the command line.

    Args:
        argv: Command-line arguments. ``None`` reads ``sys.argv``.

    Returns:
        The process exit status, which is always success once the modules
        imported and the descriptors built.
    """
    parser = ArgumentParser(
        prog="spakky-grpc-descriptor-snapshot",
        description=(
            "Dump the protobuf wire layout generated from @GrpcController "
            "classes so it can be committed and diffed in CI."
        ),
    )
    parser.add_argument(
        "modules",
        nargs="+",
        metavar="MODULE",
        help="Dotted module or package path containing @GrpcController classes.",
    )
    arguments = parser.parse_args(argv)
    # A console script starts with the installation directory on ``sys.path`` and not
    # the working directory, so without this the project the command is invoked from
    # would not be importable by name.
    sys.path.insert(0, str(Path.cwd()))
    snapshot = build_descriptor_snapshot(collect_controller_types(arguments.modules))
    print(snapshot.model_dump_json(indent=2))
    return 0

인터셉터

Tracing interceptor for W3C Trace Context propagation over gRPC.

Extracts trace context from incoming gRPC metadata, activates a child span for the RPC lifetime, and injects trace context into trailing metadata.

TracingInterceptor(*, propagator)

Bases: ServerInterceptor

Interceptor that propagates W3C Trace Context across gRPC boundaries.

Extracts traceparent / tracestate from incoming request metadata, activates a child span for the RPC lifetime, and injects the current trace context into trailing metadata.

Initialize the tracing interceptor.

Parameters:

Name	Type	Description	Default
propagator	ITracePropagator	Trace context propagator for extract/inject.	required

Source code in plugins/spakky-grpc/src/spakky/plugins/grpc/interceptors/tracing.py

def __init__(self, *, propagator: ITracePropagator) -> None:
    """Initialize the tracing interceptor.

    Args:
        propagator: Trace context propagator for extract/inject.
    """
    self.__propagator = propagator

intercept_service(continuation, handler_call_details) async

Intercept an RPC and set up W3C Trace Context.

Extracts trace context from incoming metadata, creates a child span (or a new root when no parent exists), and wraps the handler to inject trace context into trailing metadata and clear it after completion.

Parameters:

Name	Type	Description	Default
continuation	Callable[[HandlerCallDetails], Awaitable[RpcMethodHandler[RequestT, ResponseT] | None]]	Calls the next interceptor or resolves the handler.	required
handler_call_details	HandlerCallDetails	Describes the incoming RPC.	required

Returns:

Type	Description
RpcMethodHandler[RequestT, ResponseT] | None	A handler with trace-context lifecycle wrappers.

Source code in plugins/spakky-grpc/src/spakky/plugins/grpc/interceptors/tracing.py

@override
async def intercept_service[RequestT, ResponseT](
    self,
    continuation: Callable[
        [grpc.HandlerCallDetails],
        Awaitable[grpc.RpcMethodHandler[RequestT, ResponseT] | None],
    ],
    handler_call_details: grpc.HandlerCallDetails,
) -> grpc.RpcMethodHandler[RequestT, ResponseT] | None:
    """Intercept an RPC and set up W3C Trace Context.

    Extracts trace context from incoming metadata, creates a child span
    (or a new root when no parent exists), and wraps the handler to inject
    trace context into trailing metadata and clear it after completion.

    Args:
        continuation: Calls the next interceptor or resolves the handler.
        handler_call_details: Describes the incoming RPC.

    Returns:
        A handler with trace-context lifecycle wrappers.
    """
    carrier = self._metadata_to_dict(handler_call_details.invocation_metadata)
    parent = self.__propagator.extract(carrier)
    ctx = parent.child() if parent is not None else TraceContext.new_root()
    TraceContext.set(ctx)

    try:
        handler = await continuation(handler_call_details)
    except Exception:
        TraceContext.clear()
        raise

    if handler is None:
        TraceContext.clear()
        return handler

    return _WrappedHandler(
        handler,
        wrap_unary=self._wrap_unary_behavior,
        wrap_stream=self._wrap_stream_behavior,
    )

Error handling interceptor for gRPC servers.

Catches domain exceptions and maps them to appropriate gRPC status codes. Unexpected exceptions are logged and returned as INTERNAL status.

ErrorHandlingInterceptor(*, debug=False)

Bases: ServerInterceptor

Interceptor that converts exceptions to gRPC status codes.

AbstractGrpcStatusError subclasses are mapped to their declared status_code. All other exceptions become INTERNAL.

Attributes:

Name	Type	Description
__debug	bool	When True, include tracebacks in error details.

Initialize the error handling interceptor.

Parameters:

Name	Type	Description	Default
debug	bool	Whether to include full tracebacks in error details.	False

Source code in plugins/spakky-grpc/src/spakky/plugins/grpc/interceptors/error_handling.py

def __init__(self, *, debug: bool = False) -> None:
    """Initialize the error handling interceptor.

    Args:
        debug: Whether to include full tracebacks in error details.
    """
    self.__debug = debug

intercept_service(continuation, handler_call_details) async

Intercept an RPC and wrap the handler with error handling.

Parameters:

Name	Type	Description	Default
continuation	Callable[[HandlerCallDetails], Awaitable[RpcMethodHandler[RequestT, ResponseT] | None]]	Calls the next interceptor or resolves the handler.	required
handler_call_details	HandlerCallDetails	Describes the incoming RPC.	required

Returns:

Type	Description
RpcMethodHandler[RequestT, ResponseT] | None	A handler with error-catching wrappers on its behavior methods.

Source code in plugins/spakky-grpc/src/spakky/plugins/grpc/interceptors/error_handling.py

@override
async def intercept_service[RequestT, ResponseT](
    self,
    continuation: Callable[
        [grpc.HandlerCallDetails],
        Awaitable[grpc.RpcMethodHandler[RequestT, ResponseT] | None],
    ],
    handler_call_details: grpc.HandlerCallDetails,
) -> grpc.RpcMethodHandler[RequestT, ResponseT] | None:
    """Intercept an RPC and wrap the handler with error handling.

    Args:
        continuation: Calls the next interceptor or resolves the handler.
        handler_call_details: Describes the incoming RPC.

    Returns:
        A handler with error-catching wrappers on its behavior methods.
    """
    handler = await continuation(handler_call_details)
    if handler is None:
        return handler
    return _WrappedHandler(
        handler,
        wrap_unary=self._wrap_unary_behavior,
        wrap_stream=self._wrap_stream_behavior,
    )

후처리기s

Post-processor for registering gRPC services from controllers.

Scans @GrpcController-decorated Pods, builds protobuf descriptors at runtime, and appends generic RPC handlers to the shared :class:GrpcServerSpec.

RegisterServicesPostProcessor

Bases: IPostProcessor, IContainerAware, IApplicationContextAware

Post-processor that registers gRPC services from controllers.

When a @GrpcController Pod is created, this processor:

1. Builds a FileDescriptorProto from the controller's @rpc methods and dataclass message types.
2. Registers the descriptor in the shared DescriptorRegistry.
3. Creates a GrpcServiceHandler (generic handler) and appends it to the shared :class:GrpcServerSpec.

Runs at @Order(0) — first in the gRPC post-processor chain.

set_container(container)

Inject the IoC container.

Parameters:

Name	Type	Description	Default
container	IContainer	The IoC container.	required

Source code in plugins/spakky-grpc/src/spakky/plugins/grpc/post_processors/register_services.py

@override
def set_container(self, container: IContainer) -> None:
    """Inject the IoC container.

    Args:
        container: The IoC container.
    """
    self.__container = container

set_application_context(application_context)

Inject the application context.

Parameters:

Name	Type	Description	Default
application_context	IApplicationContext	The application context.	required

Source code in plugins/spakky-grpc/src/spakky/plugins/grpc/post_processors/register_services.py

@override
def set_application_context(self, application_context: IApplicationContext) -> None:
    """Inject the application context.

    Args:
        application_context: The application context.
    """
    self.__application_context = application_context

post_process(pod)

Register a gRPC service if pod is a @GrpcController.

Non-controller Pods are returned unchanged.

Parameters:

Name	Type	Description	Default
pod	object	The Pod instance to process.	required

Returns:

Type	Description
object	The unmodified Pod.

Source code in plugins/spakky-grpc/src/spakky/plugins/grpc/post_processors/register_services.py

@override
def post_process(self, pod: object) -> object:
    """Register a gRPC service if *pod* is a ``@GrpcController``.

    Non-controller Pods are returned unchanged.

    Args:
        pod: The Pod instance to process.

    Returns:
        The unmodified Pod.
    """
    if not GrpcController.exists(type(pod)):
        return pod

    controller_type = self._unwrap_proxy_type(type(pod))
    annotation = GrpcController.get(controller_type)
    package = annotation.package
    service_name = annotation.service_name or controller_type.__name__

    file_desc = build_file_descriptor(controller_type)

    registry = self.__container.get(DescriptorRegistry)
    if not registry.is_registered(file_desc.name):
        registry.register(file_desc)

    handler = GrpcServiceHandler(
        controller_type=controller_type,
        package=package,
        service_name=service_name,
        container=self.__container,
        application_context=self.__application_context,
        registry=registry,
    )

    spec = self.__container.get(GrpcServerSpec)
    spec.add_handler(handler)

    logger.info(
        f"Registered gRPC service {package}.{service_name} "
        f"from {controller_type.__qualname__}"
    )
    return pod

Post-processor for recording interceptors on the gRPC server spec.

Adds ErrorHandlingInterceptor and (when the tracing plugin is loaded) TracingInterceptor to the shared :class:GrpcServerSpec. The actual grpc.aio.Server is instantiated later, on the event loop that will run it (see :mod:spakky.plugins.grpc.server_spec).

AddInterceptorsPostProcessor

Bases: IPostProcessor, IContainerAware, IApplicationContextAware

Post-processor that records interceptors on the shared server spec.

Interceptors added (in order):

1. ErrorHandlingInterceptor — always.
2. TracingInterceptor — only when an ITracePropagator is available in the application context.

Runs at @Order(1) — after service registration.

set_container(container)

Inject the IoC container.

Parameters:

Name	Type	Description	Default
container	IContainer	The IoC container.	required

Source code in plugins/spakky-grpc/src/spakky/plugins/grpc/post_processors/add_interceptors.py

@override
def set_container(self, container: IContainer) -> None:
    """Inject the IoC container.

    Args:
        container: The IoC container.
    """
    self.__container = container

set_application_context(application_context)

Inject the application context.

Parameters:

Name	Type	Description	Default
application_context	IApplicationContext	The application context.	required

Source code in plugins/spakky-grpc/src/spakky/plugins/grpc/post_processors/add_interceptors.py

@override
def set_application_context(self, application_context: IApplicationContext) -> None:
    """Inject the application context.

    Args:
        application_context: The application context.
    """
    self.__application_context = application_context

post_process(pod)

Record interceptors on the server spec once per spec instance.

The spec is resolved lazily so that interceptor registration runs exactly once: the first time a GrpcServerSpec Pod is seen.

Parameters:

Name	Type	Description	Default
pod	object	The Pod instance to process.	required

Returns:

Type	Description
object	The pod unchanged.

Source code in plugins/spakky-grpc/src/spakky/plugins/grpc/post_processors/add_interceptors.py

@override
def post_process(self, pod: object) -> object:
    """Record interceptors on the server spec once per spec instance.

    The spec is resolved lazily so that interceptor registration runs
    exactly once: the first time a ``GrpcServerSpec`` Pod is seen.

    Args:
        pod: The Pod instance to process.

    Returns:
        The pod unchanged.
    """
    if not isinstance(pod, GrpcServerSpec):
        return pod

    pod.add_interceptor(ErrorHandlingInterceptor())

    propagator = self.__application_context.get_or_none(ITracePropagator)
    if propagator is not None:
        pod.add_interceptor(TracingInterceptor(propagator=propagator))

    logger.info(f"Registered {len(pod.interceptors)} interceptor(s) on gRPC spec")
    return pod

Post-processor for binding gRPC server lifecycle.

Wires a :class:GrpcServerSpec Pod into the ApplicationContext so that the underlying grpc.aio.Server is materialised on the context's event loop and started/stopped alongside the application.

GRACEFUL_SHUTDOWN_SECONDS = 5.0 module-attribute

Default grace period (seconds) for server shutdown.

GrpcServerService(spec)

Bases: IAsyncService

Async service wrapper that instantiates the gRPC server on the right loop.

grpc.aio.server() binds to whatever event loop is running when it is called, so the real server is created inside :meth:start_async from the captured :class:GrpcServerSpec.

Attributes:

Name	Type	Description
_spec	GrpcServerSpec	Configuration collected during post-processing.
_server	Server | None	The materialised server, set once :meth:start_async runs.
_stop_event	Event	Async event passed by the application context; unused internally because shutdown is driven by :meth:stop_async, but retained to satisfy the :class:IAsyncService contract.

Initialise the service with a server spec.

Parameters:

Name	Type	Description	Default
spec	GrpcServerSpec	Collected server configuration.	required

Source code in plugins/spakky-grpc/src/spakky/plugins/grpc/post_processors/bind_server.py

def __init__(self, spec: GrpcServerSpec) -> None:
    """Initialise the service with a server spec.

    Args:
        spec: Collected server configuration.
    """
    self._spec = spec
    self._server = None

set_stop_event(stop_event)

Store the async stop event from the application context.

Parameters:

Name	Type	Description	Default
stop_event	Event	Async event forwarded by the application context.	required

Source code in plugins/spakky-grpc/src/spakky/plugins/grpc/post_processors/bind_server.py

@override
def set_stop_event(self, stop_event: locks.Event) -> None:
    """Store the async stop event from the application context.

    Args:
        stop_event: Async event forwarded by the application context.
    """
    self._stop_event = stop_event

start_async() async

Build the gRPC server on the current loop and start it.

Source code in plugins/spakky-grpc/src/spakky/plugins/grpc/post_processors/bind_server.py

@override
async def start_async(self) -> None:
    """Build the gRPC server on the current loop and start it."""
    self._server = await self._spec.build_async()
    await self._server.start()
    logger.info("gRPC server started")

stop_async() async

Gracefully stop the gRPC server if it was started.

Clears _server after a successful stop so that repeated stop_async calls are safe no-ops.

Source code in plugins/spakky-grpc/src/spakky/plugins/grpc/post_processors/bind_server.py

@override
async def stop_async(self) -> None:
    """Gracefully stop the gRPC server if it was started.

    Clears ``_server`` after a successful stop so that repeated
    ``stop_async`` calls are safe no-ops.
    """
    if self._server is None:
        return
    server = self._server
    self._server = None
    await server.stop(grace=GRACEFUL_SHUTDOWN_SECONDS)
    logger.info("gRPC server stopped")

BindServerPostProcessor

Bases: IPostProcessor, IContainerAware, IApplicationContextAware

Post-processor that binds a :class:GrpcServerSpec to the ApplicationContext.

Wraps the spec in a :class:GrpcServerService and registers it with the ApplicationContext for automatic start/stop management.

Runs at @Order(2) — last in the gRPC post-processor chain.

set_container(container)

Inject the IoC container.

Parameters:

Name	Type	Description	Default
container	IContainer	The IoC container.	required

Source code in plugins/spakky-grpc/src/spakky/plugins/grpc/post_processors/bind_server.py

@override
def set_container(self, container: IContainer) -> None:
    """Inject the IoC container.

    Args:
        container: The IoC container.
    """
    self.__container = container

set_application_context(application_context)

Inject the application context.

Parameters:

Name	Type	Description	Default
application_context	IApplicationContext	The application context.	required

Source code in plugins/spakky-grpc/src/spakky/plugins/grpc/post_processors/bind_server.py

@override
def set_application_context(self, application_context: IApplicationContext) -> None:
    """Inject the application context.

    Args:
        application_context: The application context.
    """
    self.__application_context = application_context

post_process(pod)

Bind server lifecycle if pod is a GrpcServerSpec.

Non-spec Pods are returned unchanged.

Parameters:

Name	Type	Description	Default
pod	object	The Pod instance to process.	required

Returns:

Type	Description
object	The unmodified spec Pod.

Source code in plugins/spakky-grpc/src/spakky/plugins/grpc/post_processors/bind_server.py

@override
def post_process(self, pod: object) -> object:
    """Bind server lifecycle if *pod* is a ``GrpcServerSpec``.

    Non-spec Pods are returned unchanged.

    Args:
        pod: The Pod instance to process.

    Returns:
        The unmodified spec Pod.
    """
    if not isinstance(pod, GrpcServerSpec):
        return pod
    if not pod.bind_addresses:
        logger.info("Skipped gRPC server lifecycle binding; no bind address set")
        return pod

    service = GrpcServerService(pod)
    service.set_stop_event(self.__application_context.task_stop_event)
    self.__application_context.add_service(service)
    logger.info("Bound gRPC server lifecycle to ApplicationContext")
    return pod

에러

gRPC plugin error hierarchy.

Provides base error classes, gRPC status-mapped errors, and schema errors.

AbstractSpakkyGrpcError

Bases: AbstractSpakkyFrameworkError, ABC

Base exception for all Spakky gRPC errors.

AbstractGrpcStatusError

Bases: AbstractSpakkyGrpcError, ABC

Base for gRPC errors that map to a specific status code.

Subclasses must define status_code to specify which gRPC status code the error maps to.

InvalidArgument

Bases: AbstractGrpcStatusError

gRPC INVALID_ARGUMENT error.

NotFound

Bases: AbstractGrpcStatusError

gRPC NOT_FOUND error.

AlreadyExists

Bases: AbstractGrpcStatusError

gRPC ALREADY_EXISTS error.

PermissionDenied

Bases: AbstractGrpcStatusError

gRPC PERMISSION_DENIED error.

Unauthenticated

Bases: AbstractGrpcStatusError

gRPC UNAUTHENTICATED error.

FailedPrecondition

Bases: AbstractGrpcStatusError

gRPC FAILED_PRECONDITION error.

Unavailable

Bases: AbstractGrpcStatusError

gRPC UNAVAILABLE error.

InternalError

Bases: AbstractGrpcStatusError

gRPC INTERNAL error.

UnsupportedFieldTypeError(field_type)

Bases: AbstractSpakkyGrpcError

Raised when a Python type cannot be mapped to a protobuf type.

Source code in plugins/spakky-grpc/src/spakky/plugins/grpc/error.py

def __init__(self, field_type: type[object]) -> None:
    super().__init__()
    self.field_type = field_type

UnsupportedResponseTypeError(value_type)

Bases: AbstractSpakkyGrpcError

Raised when a serializer receives an object it cannot encode.

The gRPC response serializer accepts either a protobuf Message (passed through verbatim) or a pydantic BaseModel (encoded via the json_format bridge). Any other type signals a controller returned an unsupported value.

Source code in plugins/spakky-grpc/src/spakky/plugins/grpc/error.py

def __init__(self, value_type: type[object]) -> None:
    super().__init__()
    self.value_type = value_type

DescriptorAlreadyRegisteredError(file_name)

Bases: AbstractSpakkyGrpcError

Raised when a FileDescriptorProto is registered more than once.

Source code in plugins/spakky-grpc/src/spakky/plugins/grpc/error.py

def __init__(self, file_name: str) -> None:
    super().__init__()
    self.file_name = file_name

ProtoFieldNumberConflictError(model_type, explicit_field_name, derived_field_name, number)

Bases: AbstractSpakkyGrpcError

Raised when an explicit ProtoField number collides with an auto-derived one.

An explicit ProtoField(number=N) reserves N for its field. If an auto-numbered field in the same message natively hashes to N, silently re-hashing the auto field would change its wire number and break compatibility. This conflict is surfaced as a build error instead so the author pins the auto field's number too or chooses a different N.

Source code in plugins/spakky-grpc/src/spakky/plugins/grpc/error.py

def __init__(
    self,
    model_type: type,
    explicit_field_name: str,
    derived_field_name: str,
    number: int,
) -> None:
    super().__init__()
    self.model_type = model_type
    self.explicit_field_name = explicit_field_name
    self.derived_field_name = derived_field_name
    self.number = number

InvalidProtoFieldNumberError(model_type, field_name, number)

Bases: AbstractSpakkyGrpcError

Raised when an explicit ProtoField number is not a valid protobuf number.

An explicit ProtoField(number=N) must fall in the assignable protobuf range: 1 .. 536_870_911 (2**29 - 1) and outside the protobuf reserved band 19_000 .. 19_999. The auto-numbering path already honors these constraints deterministically, but an explicit override bypassed them and only failed later at descriptor-pool build time with an opaque protobuf message. This error surfaces the violation at schema-build time with the offending field and number so the author corrects it.

Source code in plugins/spakky-grpc/src/spakky/plugins/grpc/error.py

def __init__(self, model_type: type, field_name: str, number: int) -> None:
    super().__init__()
    self.model_type = model_type
    self.field_name = field_name
    self.number = number

DuplicateProtoFieldNumberError(model_type, first_field_name, second_field_name, number)

Bases: AbstractSpakkyGrpcError

Raised when two explicit ProtoField numbers collide in one message.

Each protobuf field number must be unique within its message. Two fields carrying the same explicit ProtoField(number=N) would later be rejected by the descriptor pool with an opaque message. This error surfaces the duplicate at schema-build time, naming both colliding fields and the shared number so the author repins one of them.

Source code in plugins/spakky-grpc/src/spakky/plugins/grpc/error.py

def __init__(
    self,
    model_type: type,
    first_field_name: str,
    second_field_name: str,
    number: int,
) -> None:
    super().__init__()
    self.model_type = model_type
    self.first_field_name = first_field_name
    self.second_field_name = second_field_name
    self.number = number

IncompleteTlsCredentialsError(certificate_chain_file, private_key_file)

Bases: AbstractSpakkyGrpcError

Raised when TLS is configured with only one half of the key pair.

Terminating TLS needs both the server certificate chain and the matching private key. Configuring one without the other would otherwise fall back to a plaintext listener, silently downgrading transport security on a deployment that asked for TLS.

Source code in plugins/spakky-grpc/src/spakky/plugins/grpc/error.py

def __init__(
    self,
    certificate_chain_file: Path | None,
    private_key_file: Path | None,
) -> None:
    super().__init__()
    self.certificate_chain_file = certificate_chain_file
    self.private_key_file = private_key_file

MissingClientCertificateAuthorityError

Bases: AbstractSpakkyGrpcError

Raised when mutual TLS is requested without a client CA to verify against.

require_client_auth makes the server reject peers whose certificate is not signed by a trusted authority. Without a client CA file there is no authority to check against, so the setting could not be honoured.

NotAnRpcMethodError(method_name)

Bases: AbstractSpakkyGrpcError

Raised when a client callable is built from a method without @rpc.

Source code in plugins/spakky-grpc/src/spakky/plugins/grpc/error.py

def __init__(self, method_name: str) -> None:
    super().__init__()
    self.method_name = method_name

RpcMethodTypeMismatchError(method_name, expected, actual)

Bases: AbstractSpakkyGrpcError

Raised when a client callable is built for the wrong streaming pattern.

Each GrpcClient factory produces a multicallable of one specific gRPC streaming shape. Building a unary callable for a server-streaming method would otherwise fail deep inside the transport with an opaque error.

Source code in plugins/spakky-grpc/src/spakky/plugins/grpc/error.py

def __init__(
    self,
    method_name: str,
    expected: RpcMethodType,
    actual: RpcMethodType,
) -> None:
    super().__init__()
    self.method_name = method_name
    self.expected = expected
    self.actual = actual

MessagelessRpcMethodError(method_name)

Bases: AbstractSpakkyGrpcError

Raised when an @rpc method declares no request or no response model.

protobuf requires every method to name an input and an output message, so a method missing either cannot be addressed over the wire. Reporting it here names the offending method instead of surfacing an opaque descriptor-pool rejection about an unresolvable empty type name.

Raised from two places that reach the same invalid declaration: controller registration (build_service_descriptor) and client callable construction. Both carry method_name as <controller>.<method> so a log line identifies the declaration even when several controllers share a method name.

Source code in plugins/spakky-grpc/src/spakky/plugins/grpc/error.py

def __init__(self, method_name: str) -> None:
    super().__init__()
    self.method_name = method_name

NoControllerFoundError(module_names)

Bases: AbstractSpakkyGrpcError

Raised when a descriptor snapshot is requested for modules with no controller.

An empty snapshot looks like a valid baseline: committing it and comparing later runs against it passes forever, so the wire layout the snapshot exists to protect goes unguarded. Naming the scanned modules instead points at the usual cause, a wrong module path on the command line.

Source code in plugins/spakky-grpc/src/spakky/plugins/grpc/error.py

def __init__(self, module_names: tuple[str, ...]) -> None:
    super().__init__()
    self.module_names = module_names

추가 모듈

Plugin initialization for gRPC integration.

Registers post-processors that enable automatic gRPC service registration, interceptor injection, and server lifecycle management.

initialize(app)

Initialize the gRPC plugin.

Registers post-processors for automatic gRPC service registration, interceptor injection, and server lifecycle management. This function is called automatically by the Spakky framework during plugin loading.

Parameters:

Name	Type	Description	Default
app	SpakkyApplication	The Spakky application instance.	required

Source code in plugins/spakky-grpc/src/spakky/plugins/grpc/main.py

def initialize(app: SpakkyApplication) -> None:
    """Initialize the gRPC plugin.

    Registers post-processors for automatic gRPC service registration,
    interceptor injection, and server lifecycle management.  This
    function is called automatically by the Spakky framework during
    plugin loading.

    Args:
        app: The Spakky application instance.
    """
    app.add(GrpcConfig)
    app.add(descriptor_registry)
    app.add(grpc_health_servicer)
    app.add(grpc_server_spec)
    app.add(RegisterServicesPostProcessor)
    app.add(AddInterceptorsPostProcessor)
    app.add(BindServerPostProcessor)

descriptor_registry()

Create the shared protobuf descriptor registry.

Source code in plugins/spakky-grpc/src/spakky/plugins/grpc/main.py

@Pod(name="descriptor_registry")
def descriptor_registry() -> DescriptorRegistry:
    """Create the shared protobuf descriptor registry."""
    return DescriptorRegistry()

grpc_health_servicer()

Create the servicer backing the standard health checking service.

Applications inject this Pod to report their own serving status, for example flipping a service to NOT_SERVING while a dependency is down.

Source code in plugins/spakky-grpc/src/spakky/plugins/grpc/main.py

@Pod(name="grpc_health_servicer")
def grpc_health_servicer() -> health.aio.HealthServicer:
    """Create the servicer backing the standard health checking service.

    Applications inject this Pod to report their own serving status, for
    example flipping a service to ``NOT_SERVING`` while a dependency is down.
    """
    return health.aio.HealthServicer()

grpc_server_spec(config, descriptor_registry, grpc_health_servicer)

Create the shared gRPC server spec from plugin configuration.

Parameters:

Name	Type	Description	Default
config	GrpcConfig	Plugin configuration.	required
descriptor_registry	DescriptorRegistry	Registry the standard services publish into.	required
grpc_health_servicer	HealthServicer	Servicer holding the reported serving statuses.	required

Returns:

Type	Description
GrpcServerSpec	The spec collecting everything the server is built from.

Source code in plugins/spakky-grpc/src/spakky/plugins/grpc/main.py

@Pod(name="grpc_server_spec")
def grpc_server_spec(
    config: GrpcConfig,
    descriptor_registry: DescriptorRegistry,
    grpc_health_servicer: health.aio.HealthServicer,
) -> GrpcServerSpec:
    """Create the shared gRPC server spec from plugin configuration.

    Args:
        config: Plugin configuration.
        descriptor_registry: Registry the standard services publish into.
        grpc_health_servicer: Servicer holding the reported serving statuses.

    Returns:
        The spec collecting everything the server is built from.
    """
    spec = GrpcServerSpec(options=config.server_options)
    credentials = build_server_credentials(config)
    for address in config.bind_addresses:
        if credentials is None:
            spec.add_insecure_port(address)
        else:
            spec.add_secure_port(address, credentials)
    if config.health_service_enabled:
        enable_health_service(spec, descriptor_registry, grpc_health_servicer)
    if config.reflection_service_enabled:
        enable_reflection_service(spec, descriptor_registry)
    return spec