Source code for viam.rpc.types

from abc import abstractmethod
from typing import Mapping, Protocol, runtime_checkable

import grpclib
from grpclib._typing import IServable
from grpclib.client import Channel


class RPCServiceBase(IServable):
    """The base requirements for an RPC Service.

    An RPC Service is a service that can handle incoming RPC requests.
    Services that implement the service bases generated by grpclib will automatically conform to this protocol.
    """

    @abstractmethod
    def __mapping__(self) -> Mapping[str, grpclib.const.Handler]:
        ...


@runtime_checkable
class RPCServiceStubBase(Protocol):
    def __init__(self, channel: Channel) -> None:
        ...