text: # Generated by the protocol buffer compiler.  DO NOT EDIT!
  # sources: onos/ransim/metrics/metrics.proto
  # plugin: python-betterproto
  from dataclasses import dataclass
  from typing import AsyncIterator, Dict, List
  
  import betterproto
  from betterproto.grpc.grpclib_server import ServiceBase
  import grpclib
  
  
  class EventType(betterproto.Enum):
      """Change event type"""
  
      # NONE indicates unknown event type
      NONE = 0
      # UPDATED indicates a metric value was set (updated)
      UPDATED = 1
      # DELETED indicates a metric was deleted
      DELETED = 2
  
  
  @dataclass(eq=False, repr=False)
  class Metric(betterproto.Message):
      entityid: int = betterproto.uint64_field(1)
      key: str = betterproto.string_field(2)
      value: str = betterproto.string_field(3)
      type: str = betterproto.string_field(4)
  
  
  @dataclass(eq=False, repr=False)
  class ListRequest(betterproto.Message):
      entityid: int = betterproto.uint64_field(1)
  
  
  @dataclass(eq=False, repr=False)
  class ListResponse(betterproto.Message):
      entityid: int = betterproto.uint64_field(1)
      metrics: List["Metric"] = betterproto.message_field(2)
  
  
  @dataclass(eq=False, repr=False)
  class SetRequest(betterproto.Message):
      metric: "Metric" = betterproto.message_field(1)
  
  
  @dataclass(eq=False, repr=False)
  class SetResponse(betterproto.Message):
      pass
  
  
  @dataclass(eq=False, repr=False)
  class GetRequest(betterproto.Message):
      entityid: int = betterproto.uint64_field(1)
      name: str = betterproto.string_field(2)
  
  
  @dataclass(eq=False, repr=False)
  class GetResponse(betterproto.Message):
      metric: "Metric" = betterproto.message_field(1)
  
  
  @dataclass(eq=False, repr=False)
  class DeleteRequest(betterproto.Message):
      entityid: int = betterproto.uint64_field(1)
      name: str = betterproto.string_field(2)
  
  
  @dataclass(eq=False, repr=False)
  class DeleteResponse(betterproto.Message):
      pass
  
  
  @dataclass(eq=False, repr=False)
  class DeleteAllRequest(betterproto.Message):
      entityid: int = betterproto.uint64_field(1)
  
  
  @dataclass(eq=False, repr=False)
  class DeleteAllResponse(betterproto.Message):
      pass
  
  
  @dataclass(eq=False, repr=False)
  class WatchRequest(betterproto.Message):
      pass
  
  
  @dataclass(eq=False, repr=False)
  class WatchResponse(betterproto.Message):
      metric: "Metric" = betterproto.message_field(1)
      type: "EventType" = betterproto.enum_field(2)
  
  
  class MetricsServiceStub(betterproto.ServiceStub):
      async def list(self, *, entityid: int = 0) -> "ListResponse":
  
          request = ListRequest()
          request.entityid = entityid
  
          return await self._unary_unary(
              "/onos.ransim.metrics.MetricsService/List", request, ListResponse
          )
  
      async def set(self, *, metric: "Metric" = None) -> "SetResponse":
  
          request = SetRequest()
          if metric is not None:
              request.metric = metric
  
          return await self._unary_unary(
              "/onos.ransim.metrics.MetricsService/Set", request, SetResponse
          )
  
      async def get(self, *, entityid: int = 0, name: str = "") -> "GetResponse":
  
          request = GetRequest()
          request.entityid = entityid
          request.name = name
  
          return await self._unary_unary(
              "/onos.ransim.metrics.MetricsService/Get", request, GetResponse
          )
  
      async def delete(self, *, entityid: int = 0, name: str = "") -> "DeleteResponse":
  
          request = DeleteRequest()
          request.entityid = entityid
          request.name = name
  
          return await self._unary_unary(
              "/onos.ransim.metrics.MetricsService/Delete", request, DeleteResponse
          )
  
      async def delete_all(self, *, entityid: int = 0) -> "DeleteAllResponse":
  
          request = DeleteAllRequest()
          request.entityid = entityid
  
          return await self._unary_unary(
              "/onos.ransim.metrics.MetricsService/DeleteAll", request, DeleteAllResponse
          )
  
      async def watch(self) -> AsyncIterator["WatchResponse"]:
  
          request = WatchRequest()
  
          async for response in self._unary_stream(
              "/onos.ransim.metrics.MetricsService/Watch",
              request,
              WatchResponse,
          ):
              yield response
  
  
  class MetricsServiceBase(ServiceBase):
      async def list(self, entityid: int) -> "ListResponse":
          raise grpclib.GRPCError(grpclib.const.Status.UNIMPLEMENTED)
  
      async def set(self, metric: "Metric") -> "SetResponse":
          raise grpclib.GRPCError(grpclib.const.Status.UNIMPLEMENTED)
  
      async def get(self, entityid: int, name: str) -> "GetResponse":
          raise grpclib.GRPCError(grpclib.const.Status.UNIMPLEMENTED)
  
      async def delete(self, entityid: int, name: str) -> "DeleteResponse":
          raise grpclib.GRPCError(grpclib.const.Status.UNIMPLEMENTED)
  
      async def delete_all(self, entityid: int) -> "DeleteAllResponse":
          raise grpclib.GRPCError(grpclib.const.Status.UNIMPLEMENTED)
  
      async def watch(self) -> AsyncIterator["WatchResponse"]:
          raise grpclib.GRPCError(grpclib.const.Status.UNIMPLEMENTED)
  
      async def __rpc_list(self, stream: grpclib.server.Stream) -> None:
          request = await stream.recv_message()
  
          request_kwargs = {
              "entityid": request.entityid,
          }
  
          response = await self.list(**request_kwargs)
          await stream.send_message(response)
  
      async def __rpc_set(self, stream: grpclib.server.Stream) -> None:
          request = await stream.recv_message()
  
          request_kwargs = {
              "metric": request.metric,
          }
  
          response = await self.set(**request_kwargs)
          await stream.send_message(response)
  
      async def __rpc_get(self, stream: grpclib.server.Stream) -> None:
          request = await stream.recv_message()
  
          request_kwargs = {
              "entityid": request.entityid,
              "name": request.name,
          }
  
          response = await self.get(**request_kwargs)
          await stream.send_message(response)
  
      async def __rpc_delete(self, stream: grpclib.server.Stream) -> None:
          request = await stream.recv_message()
  
          request_kwargs = {
              "entityid": request.entityid,
              "name": request.name,
          }
  
          response = await self.delete(**request_kwargs)
          await stream.send_message(response)
  
      async def __rpc_delete_all(self, stream: grpclib.server.Stream) -> None:
          request = await stream.recv_message()
  
          request_kwargs = {
              "entityid": request.entityid,
          }
  
          response = await self.delete_all(**request_kwargs)
          await stream.send_message(response)
  
      async def __rpc_watch(self, stream: grpclib.server.Stream) -> None:
          request = await stream.recv_message()
  
          request_kwargs = {}
  
          await self._call_rpc_handler_server_stream(
              self.watch,
              stream,
              request_kwargs,
          )
  
      def __mapping__(self) -> Dict[str, grpclib.const.Handler]:
          return {
              "/onos.ransim.metrics.MetricsService/List": grpclib.const.Handler(
                  self.__rpc_list,
                  grpclib.const.Cardinality.UNARY_UNARY,
                  ListRequest,
                  ListResponse,
              ),
              "/onos.ransim.metrics.MetricsService/Set": grpclib.const.Handler(
                  self.__rpc_set,
                  grpclib.const.Cardinality.UNARY_UNARY,
                  SetRequest,
                  SetResponse,
              ),
              "/onos.ransim.metrics.MetricsService/Get": grpclib.const.Handler(
                  self.__rpc_get,
                  grpclib.const.Cardinality.UNARY_UNARY,
                  GetRequest,
                  GetResponse,
              ),
              "/onos.ransim.metrics.MetricsService/Delete": grpclib.const.Handler(
                  self.__rpc_delete,
                  grpclib.const.Cardinality.UNARY_UNARY,
                  DeleteRequest,
                  DeleteResponse,
              ),
              "/onos.ransim.metrics.MetricsService/DeleteAll": grpclib.const.Handler(
                  self.__rpc_delete_all,
                  grpclib.const.Cardinality.UNARY_UNARY,
                  DeleteAllRequest,
                  DeleteAllResponse,
              ),
              "/onos.ransim.metrics.MetricsService/Watch": grpclib.const.Handler(
                  self.__rpc_watch,
                  grpclib.const.Cardinality.UNARY_STREAM,
                  WatchRequest,
                  WatchResponse,
              ),
          }
  