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

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


@dataclass(eq=False, repr=False)
class ListenRequest(betterproto.Message):
    channel: str = betterproto.string_field(1)


@dataclass(eq=False, repr=False)
class ListenResponse(betterproto.Message):
    extra: str = betterproto.string_field(1)


class ApiStub(betterproto.ServiceStub):
    async def listen(self, *, channel: str = "") -> AsyncIterator["ListenResponse"]:

        request = ListenRequest()
        request.channel = channel

        async for response in self._unary_stream(
            "/proxy.API/Listen",
            request,
            ListenResponse,
        ):
            yield response


class ApiBase(ServiceBase):
    async def listen(self, channel: str) -> AsyncIterator["ListenResponse"]:
        raise grpclib.GRPCError(grpclib.const.Status.UNIMPLEMENTED)

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

        request_kwargs = {
            "channel": request.channel,
        }

        await self._call_rpc_handler_server_stream(
            self.listen,
            stream,
            request_kwargs,
        )

    def __mapping__(self) -> Dict[str, grpclib.const.Handler]:
        return {
            "/proxy.API/Listen": grpclib.const.Handler(
                self.__rpc_listen,
                grpclib.const.Cardinality.UNARY_STREAM,
                ListenRequest,
                ListenResponse,
            ),
        }
