text: # Generated by the protocol buffer compiler.  DO NOT EDIT!
  # sources: atomix/value/value.proto
  # plugin: python-betterproto
  from dataclasses import dataclass
  from typing import AsyncGenerator, Optional
  
  import betterproto
  
  from atomix.proto import headers
  
  
  class EventResponseType(betterproto.Enum):
      UPDATED = 0
  
  
  @dataclass
  class CreateRequest(betterproto.Message):
      header: headers.RequestHeader = betterproto.message_field(1)
  
  
  @dataclass
  class CreateResponse(betterproto.Message):
      header: headers.ResponseHeader = betterproto.message_field(1)
  
  
  @dataclass
  class CloseRequest(betterproto.Message):
      header: headers.RequestHeader = betterproto.message_field(1)
      delete: bool = betterproto.bool_field(2)
  
  
  @dataclass
  class CloseResponse(betterproto.Message):
      header: headers.ResponseHeader = betterproto.message_field(1)
  
  
  @dataclass
  class GetRequest(betterproto.Message):
      header: headers.RequestHeader = betterproto.message_field(1)
  
  
  @dataclass
  class GetResponse(betterproto.Message):
      header: headers.ResponseHeader = betterproto.message_field(1)
      value: bytes = betterproto.bytes_field(2)
      version: int = betterproto.uint64_field(3)
  
  
  @dataclass
  class SetRequest(betterproto.Message):
      header: headers.RequestHeader = betterproto.message_field(1)
      expect_version: int = betterproto.uint64_field(2)
      expect_value: bytes = betterproto.bytes_field(3)
      value: bytes = betterproto.bytes_field(4)
  
  
  @dataclass
  class SetResponse(betterproto.Message):
      header: headers.ResponseHeader = betterproto.message_field(1)
      version: int = betterproto.uint64_field(2)
      succeeded: bool = betterproto.bool_field(3)
  
  
  @dataclass
  class EventRequest(betterproto.Message):
      header: headers.RequestHeader = betterproto.message_field(1)
  
  
  @dataclass
  class EventResponse(betterproto.Message):
      header: headers.ResponseHeader = betterproto.message_field(1)
      type: "EventResponseType" = betterproto.enum_field(2)
      previous_value: bytes = betterproto.bytes_field(3)
      previous_version: int = betterproto.uint64_field(4)
      new_value: bytes = betterproto.bytes_field(5)
      new_version: int = betterproto.uint64_field(6)
  
  
  class ValueServiceStub(betterproto.ServiceStub):
      """ValueService implements a distributed atomic value"""
  
      async def create(
          self, *, header: Optional[headers.RequestHeader] = None
      ) -> CreateResponse:
          """Create creates a new value session"""
  
          request = CreateRequest()
          if header is not None:
              request.header = header
  
          return await self._unary_unary(
              "/atomix.value.ValueService/Create", request, CreateResponse,
          )
  
      async def close(
          self, *, header: Optional[headers.RequestHeader] = None, delete: bool = False
      ) -> CloseResponse:
          """Close closes the value session"""
  
          request = CloseRequest()
          if header is not None:
              request.header = header
          request.delete = delete
  
          return await self._unary_unary(
              "/atomix.value.ValueService/Close", request, CloseResponse,
          )
  
      async def set(
          self,
          *,
          header: Optional[headers.RequestHeader] = None,
          expect_version: int = 0,
          expect_value: bytes = b"",
          value: bytes = b"",
      ) -> SetResponse:
          """Set sets the value"""
  
          request = SetRequest()
          if header is not None:
              request.header = header
          request.expect_version = expect_version
          request.expect_value = expect_value
          request.value = value
  
          return await self._unary_unary(
              "/atomix.value.ValueService/Set", request, SetResponse,
          )
  
      async def get(
          self, *, header: Optional[headers.RequestHeader] = None
      ) -> GetResponse:
          """Get gets the value"""
  
          request = GetRequest()
          if header is not None:
              request.header = header
  
          return await self._unary_unary(
              "/atomix.value.ValueService/Get", request, GetResponse,
          )
  
      async def events(
          self, *, header: Optional[headers.RequestHeader] = None
      ) -> AsyncGenerator[EventResponse, None]:
          """Events listens for value change events"""
  
          request = EventRequest()
          if header is not None:
              request.header = header
  
          async for response in self._unary_stream(
              "/atomix.value.ValueService/Events", request, EventResponse,
          ):
              yield response
  