Source code for python_pachyderm.experimental.proto.v2.versionpb_v2

# Generated by the protocol buffer compiler.  DO NOT EDIT!
# sources: python_pachyderm/proto/v2/version/versionpb/version.proto
# plugin: python-betterproto
from dataclasses import dataclass
from typing import Dict

import betterproto
from betterproto.grpc.grpclib_server import ServiceBase
import grpclib


@dataclass(eq=False, repr=False)
class Version(betterproto.Message):
    major: int = betterproto.uint32_field(1)
    minor: int = betterproto.uint32_field(2)
    micro: int = betterproto.uint32_field(3)
    additional: str = betterproto.string_field(4)


class ApiStub(betterproto.ServiceStub):
    async def get_version(self) -> "Version":

        request = betterproto_lib_google_protobuf.Empty()

        return await self._unary_unary("/versionpb_v2.API/GetVersion", request, Version)


class ApiBase(ServiceBase):
    async def get_version(self) -> "Version":
        raise grpclib.GRPCError(grpclib.const.Status.UNIMPLEMENTED)

    async def __rpc_get_version(self, stream: grpclib.server.Stream) -> None:
        request = await stream.recv_message()

        request_kwargs = {}

        response = await self.get_version(**request_kwargs)
        await stream.send_message(response)

    def __mapping__(self) -> Dict[str, grpclib.const.Handler]:
        return {
            "/versionpb_v2.API/GetVersion": grpclib.const.Handler(
                self.__rpc_get_version,
                grpclib.const.Cardinality.UNARY_UNARY,
                betterproto_lib_google_protobuf.Empty,
                Version,
            ),
        }


import betterproto.lib.google.protobuf as betterproto_lib_google_protobuf