text: # Generated by the protocol buffer compiler.  DO NOT EDIT!
  # sources: atomix/primitive/indexedmap/v1/manager.proto, atomix/primitive/indexedmap/v1/primitive.proto
  # plugin: python-betterproto
  from dataclasses import dataclass
  from datetime import timedelta
  from typing import AsyncIterator, List, Optional
  
  import betterproto
  import grpclib
  
  
  class IndexedMapCacheStrategy(betterproto.Enum):
      NEAR = 0
      READ_THROUGH = 1
      WRITE_THROUGH = 2
      READ_THROUGH_WRITE_THROUGH = 3
  
  
  class EventType(betterproto.Enum):
      NONE = 0
      INSERT = 1
      UPDATE = 2
      REMOVE = 3
      REPLAY = 4
  
  
  @dataclass(eq=False, repr=False)
  class OpenSessionRequest(betterproto.Message):
      primitive_id: str = betterproto.string_field(1)
      options: "IndexedMapSessionOptions" = betterproto.message_field(2)
  
      def __post_init__(self) -> None:
          super().__post_init__()
  
  
  @dataclass(eq=False, repr=False)
  class OpenSessionResponse(betterproto.Message):
      session_id: int = betterproto.uint64_field(1)
  
      def __post_init__(self) -> None:
          super().__post_init__()
  
  
  @dataclass(eq=False, repr=False)
  class IndexedMapSessionOptions(betterproto.Message):
      cache: "IndexedMapCacheOptions" = betterproto.message_field(1)
  
      def __post_init__(self) -> None:
          super().__post_init__()
  
  
  @dataclass(eq=False, repr=False)
  class IndexedMapCacheOptions(betterproto.Message):
      enabled: bool = betterproto.bool_field(1)
      strategy: "IndexedMapCacheStrategy" = betterproto.enum_field(2)
  
      def __post_init__(self) -> None:
          super().__post_init__()
  
  
  @dataclass(eq=False, repr=False)
  class CloseSessionRequest(betterproto.Message):
      session_id: int = betterproto.uint64_field(1)
  
      def __post_init__(self) -> None:
          super().__post_init__()
  
  
  @dataclass(eq=False, repr=False)
  class CloseSessionResponse(betterproto.Message):
      pass
  
      def __post_init__(self) -> None:
          super().__post_init__()
  
  
  @dataclass(eq=False, repr=False)
  class SizeRequest(betterproto.Message):
      pass
  
      def __post_init__(self) -> None:
          super().__post_init__()
  
  
  @dataclass(eq=False, repr=False)
  class SizeResponse(betterproto.Message):
      size: int = betterproto.uint32_field(2)
  
      def __post_init__(self) -> None:
          super().__post_init__()
  
  
  @dataclass(eq=False, repr=False)
  class Precondition(betterproto.Message):
      metadata: "__meta_v1__.ObjectMeta" = betterproto.message_field(
          1, group="precondition"
      )
  
      def __post_init__(self) -> None:
          super().__post_init__()
  
  
  @dataclass(eq=False, repr=False)
  class PutRequest(betterproto.Message):
      entry: "Entry" = betterproto.message_field(2)
      preconditions: List["Precondition"] = betterproto.message_field(3)
  
      def __post_init__(self) -> None:
          super().__post_init__()
  
  
  @dataclass(eq=False, repr=False)
  class PutResponse(betterproto.Message):
      entry: "Entry" = betterproto.message_field(2)
  
      def __post_init__(self) -> None:
          super().__post_init__()
  
  
  @dataclass(eq=False, repr=False)
  class GetRequest(betterproto.Message):
      position: "Position" = betterproto.message_field(2)
  
      def __post_init__(self) -> None:
          super().__post_init__()
  
  
  @dataclass(eq=False, repr=False)
  class GetResponse(betterproto.Message):
      entry: "Entry" = betterproto.message_field(2)
  
      def __post_init__(self) -> None:
          super().__post_init__()
  
  
  @dataclass(eq=False, repr=False)
  class FirstEntryRequest(betterproto.Message):
      pass
  
      def __post_init__(self) -> None:
          super().__post_init__()
  
  
  @dataclass(eq=False, repr=False)
  class FirstEntryResponse(betterproto.Message):
      entry: "Entry" = betterproto.message_field(2)
  
      def __post_init__(self) -> None:
          super().__post_init__()
  
  
  @dataclass(eq=False, repr=False)
  class LastEntryRequest(betterproto.Message):
      pass
  
      def __post_init__(self) -> None:
          super().__post_init__()
  
  
  @dataclass(eq=False, repr=False)
  class LastEntryResponse(betterproto.Message):
      entry: "Entry" = betterproto.message_field(2)
  
      def __post_init__(self) -> None:
          super().__post_init__()
  
  
  @dataclass(eq=False, repr=False)
  class PrevEntryRequest(betterproto.Message):
      index: int = betterproto.uint64_field(2)
  
      def __post_init__(self) -> None:
          super().__post_init__()
  
  
  @dataclass(eq=False, repr=False)
  class PrevEntryResponse(betterproto.Message):
      entry: "Entry" = betterproto.message_field(2)
  
      def __post_init__(self) -> None:
          super().__post_init__()
  
  
  @dataclass(eq=False, repr=False)
  class NextEntryRequest(betterproto.Message):
      index: int = betterproto.uint64_field(2)
  
      def __post_init__(self) -> None:
          super().__post_init__()
  
  
  @dataclass(eq=False, repr=False)
  class NextEntryResponse(betterproto.Message):
      entry: "Entry" = betterproto.message_field(2)
  
      def __post_init__(self) -> None:
          super().__post_init__()
  
  
  @dataclass(eq=False, repr=False)
  class RemoveRequest(betterproto.Message):
      entry: "Entry" = betterproto.message_field(2)
      preconditions: List["Precondition"] = betterproto.message_field(3)
  
      def __post_init__(self) -> None:
          super().__post_init__()
  
  
  @dataclass(eq=False, repr=False)
  class RemoveResponse(betterproto.Message):
      entry: "Entry" = betterproto.message_field(2)
  
      def __post_init__(self) -> None:
          super().__post_init__()
  
  
  @dataclass(eq=False, repr=False)
  class ClearRequest(betterproto.Message):
      pass
  
      def __post_init__(self) -> None:
          super().__post_init__()
  
  
  @dataclass(eq=False, repr=False)
  class ClearResponse(betterproto.Message):
      pass
  
      def __post_init__(self) -> None:
          super().__post_init__()
  
  
  @dataclass(eq=False, repr=False)
  class EntriesRequest(betterproto.Message):
      pass
  
      def __post_init__(self) -> None:
          super().__post_init__()
  
  
  @dataclass(eq=False, repr=False)
  class EntriesResponse(betterproto.Message):
      entry: "Entry" = betterproto.message_field(2)
  
      def __post_init__(self) -> None:
          super().__post_init__()
  
  
  @dataclass(eq=False, repr=False)
  class EventsRequest(betterproto.Message):
      pos: "Position" = betterproto.message_field(2)
      replay: bool = betterproto.bool_field(3)
  
      def __post_init__(self) -> None:
          super().__post_init__()
  
  
  @dataclass(eq=False, repr=False)
  class EventsResponse(betterproto.Message):
      event: "Event" = betterproto.message_field(2)
  
      def __post_init__(self) -> None:
          super().__post_init__()
  
  
  @dataclass(eq=False, repr=False)
  class Event(betterproto.Message):
      type: "EventType" = betterproto.enum_field(1)
      entry: "Entry" = betterproto.message_field(2)
  
      def __post_init__(self) -> None:
          super().__post_init__()
  
  
  @dataclass(eq=False, repr=False)
  class Position(betterproto.Message):
      index: int = betterproto.uint64_field(1)
      key: str = betterproto.string_field(2)
  
      def __post_init__(self) -> None:
          super().__post_init__()
  
  
  @dataclass(eq=False, repr=False)
  class Value(betterproto.Message):
      meta: "__meta_v1__.ObjectMeta" = betterproto.message_field(1)
      value: bytes = betterproto.bytes_field(2)
      ttl: timedelta = betterproto.message_field(3)
  
      def __post_init__(self) -> None:
          super().__post_init__()
  
  
  @dataclass(eq=False, repr=False)
  class Entry(betterproto.Message):
      pos: "Position" = betterproto.message_field(1)
      value: "Value" = betterproto.message_field(2)
  
      def __post_init__(self) -> None:
          super().__post_init__()
  
  
  class IndexedMapManagerStub(betterproto.ServiceStub):
      """IndexedMapManager is a service for managing indexed map sessions"""
  
      async def open_session(
          self, *, primitive_id: str = "", options: "IndexedMapSessionOptions" = None
      ) -> "OpenSessionResponse":
  
          request = OpenSessionRequest()
          request.primitive_id = primitive_id
          if options is not None:
              request.options = options
  
          return await self._unary_unary(
              "/atomix.primitive.indexedmap.v1.IndexedMapManager/OpenSession",
              request,
              OpenSessionResponse,
          )
  
      async def close_session(self, *, session_id: int = 0) -> "CloseSessionResponse":
  
          request = CloseSessionRequest()
          request.session_id = session_id
  
          return await self._unary_unary(
              "/atomix.primitive.indexedmap.v1.IndexedMapManager/CloseSession",
              request,
              CloseSessionResponse,
          )
  
  
  class IndexedMapStub(betterproto.ServiceStub):
      """IndexedMapManager is a service for managing indexed map sessions"""
  
      async def size(self) -> "SizeResponse":
  
          request = SizeRequest()
  
          return await self._unary_unary(
              "/atomix.primitive.indexedmap.v1.IndexedMap/Size", request, SizeResponse
          )
  
      async def put(
          self,
          *,
          entry: "Entry" = None,
          preconditions: Optional[List["Precondition"]] = None,
      ) -> "PutResponse":
          preconditions = preconditions or []
  
          request = PutRequest()
          if entry is not None:
              request.entry = entry
          if preconditions is not None:
              request.preconditions = preconditions
  
          return await self._unary_unary(
              "/atomix.primitive.indexedmap.v1.IndexedMap/Put", request, PutResponse
          )
  
      async def get(self, *, position: "Position" = None) -> "GetResponse":
  
          request = GetRequest()
          if position is not None:
              request.position = position
  
          return await self._unary_unary(
              "/atomix.primitive.indexedmap.v1.IndexedMap/Get", request, GetResponse
          )
  
      async def first_entry(self) -> "FirstEntryResponse":
  
          request = FirstEntryRequest()
  
          return await self._unary_unary(
              "/atomix.primitive.indexedmap.v1.IndexedMap/FirstEntry",
              request,
              FirstEntryResponse,
          )
  
      async def last_entry(self) -> "LastEntryResponse":
  
          request = LastEntryRequest()
  
          return await self._unary_unary(
              "/atomix.primitive.indexedmap.v1.IndexedMap/LastEntry",
              request,
              LastEntryResponse,
          )
  
      async def prev_entry(self, *, index: int = 0) -> "PrevEntryResponse":
  
          request = PrevEntryRequest()
          request.index = index
  
          return await self._unary_unary(
              "/atomix.primitive.indexedmap.v1.IndexedMap/PrevEntry",
              request,
              PrevEntryResponse,
          )
  
      async def next_entry(self, *, index: int = 0) -> "NextEntryResponse":
  
          request = NextEntryRequest()
          request.index = index
  
          return await self._unary_unary(
              "/atomix.primitive.indexedmap.v1.IndexedMap/NextEntry",
              request,
              NextEntryResponse,
          )
  
      async def remove(
          self,
          *,
          entry: "Entry" = None,
          preconditions: Optional[List["Precondition"]] = None,
      ) -> "RemoveResponse":
          preconditions = preconditions or []
  
          request = RemoveRequest()
          if entry is not None:
              request.entry = entry
          if preconditions is not None:
              request.preconditions = preconditions
  
          return await self._unary_unary(
              "/atomix.primitive.indexedmap.v1.IndexedMap/Remove", request, RemoveResponse
          )
  
      async def clear(self) -> "ClearResponse":
  
          request = ClearRequest()
  
          return await self._unary_unary(
              "/atomix.primitive.indexedmap.v1.IndexedMap/Clear", request, ClearResponse
          )
  
      async def events(
          self, *, pos: "Position" = None, replay: bool = False
      ) -> AsyncIterator["EventsResponse"]:
  
          request = EventsRequest()
          if pos is not None:
              request.pos = pos
          request.replay = replay
  
          async for response in self._unary_stream(
              "/atomix.primitive.indexedmap.v1.IndexedMap/Events",
              request,
              EventsResponse,
          ):
              yield response
  
      async def entries(self) -> AsyncIterator["EntriesResponse"]:
  
          request = EntriesRequest()
  
          async for response in self._unary_stream(
              "/atomix.primitive.indexedmap.v1.IndexedMap/Entries",
              request,
              EntriesResponse,
          ):
              yield response
  
  
  from ...meta import v1 as __meta_v1__
  