text: import typing as tg
  import uuid
  
  import eventualpy.datatypes as dts
  import eventualpy.messages as msgs
  from eventualpy import publisher
  from eventualpy import reader
  from eventualpy import subscriptions
  from eventualpy import connection
  
  
  class Client:
      def __init__(self, base_url: str = "http://127.0.0.1:2113", loop=None):
          self._connection = connection.Connection(base_url=base_url, loop=loop)
  
      async def publish_event(
          self,
          stream: str,
          type: str,
          id: dts.EventId = None,
          body: dts.EventData = None,
          metadata: dts.EventMetaData = None,
          expected_version: int = dts.ExpectedVersion.Any,
      ):
          return await self.publish(
              stream=stream,
              new_events=[
                  msgs.NewEvent.make(type=type, id=id, data=body, metadata=metadata)
              ],
              expected_version=expected_version,
          )
  
      async def publish(
          self,
          stream: str,
          new_events: tg.Sequence[msgs.NewEvent],
          expected_version: int = dts.ExpectedVersion.Any,
      ):
          action = publisher.Publisher(
              self._connection, stream, new_events, expected_version
          )
          return await action.publish()
  
      async def get_event(self, stream: str, event_number: int, require_master=False):
          action = reader.SingleReader(
              self._connection,
              stream,
              event_number=event_number,
              require_master=require_master,
          )
          return await action.read()
  
      async def get(
          self,
          stream: str,
          direction: dts.StreamDirection = dts.StreamDirection.Forward,
          from_event: int = 0,
          max_count: int = 100,
          require_master: bool = False,
      ):
          action = reader.Reader(
              self._connection,
              link=dts.Link(stream, from_event, direction),
              max_count=max_count,
              require_master=require_master,
          )
          return await action.read()
  
      async def get_all(
          self,
          direction: dts.StreamDirection = dts.StreamDirection.Forward,
          from_position: dts.Position = dts.Position.last(),
          max_count: int = 100,
          require_master: bool = False,
      ):
          action = reader.Reader(
              self._connection,
              link=dts.Link("$all", from_position, direction),
              max_count=max_count,
              require_master=require_master,
          )
          return await action.read()
  
      async def iter(
          self,
          stream: str,
          direction: dts.StreamDirection = dts.StreamDirection.Forward,
          from_event: int = None,
          batch_size: int = 100,
          resolve_links: bool = True,
          require_master: bool = False,
      ):
          action = reader.Iterator(
              self._connection,
              link=dts.Link(stream, from_event, direction, count=batch_size),
              require_master=require_master,
          )
          async for event in action.read():
              yield event
  
      async def iter_all(
          self,
          direction: dts.StreamDirection = dts.StreamDirection.Forward,
          from_position: dts.Position = dts.Position.last(),
          batch_size: int = 100,
          resolve_links: bool = True,
          require_master: bool = False,
      ):
          action = reader.Iterator(
              self._connection,
              link=dts.Link("$all", from_position, direction, count=batch_size),
              require_master=require_master,
          )
          async for event in action.read():
              yield event
  
      async def create_subscription(
          self,
          name: str,
          stream: str,
          resolve_links: bool = True,
          start_from: int = -1,
          timeout_ms: int = 30000,
          record_statistics: bool = False,
          live_buffer_size: int = 500,
          read_batch_size: int = 500,
          buffer_size: int = 1000,
          max_retry_count: int = 10,
          checkpoint_after_ms: int = 2000,
          checkpoint_max_count: int = 1000,
          checkpoint_min_count: int = 10,
          subscriber_max_count: int = 10,
          credentials=None,
          consumer_strategy=None,
          fail_if_exists: bool = True,
      ):
          action = subscriptions.Subscription(self._connection, stream, name)
          await action.create(
              resolve_links=resolve_links,
              start_from=start_from,
              timeout_ms=timeout_ms,
              record_statistics=record_statistics,
              live_buffer_size=live_buffer_size,
              read_batch_size=read_batch_size,
              buffer_size=buffer_size,
              max_retry_count=max_retry_count,
              checkpoint_after_ms=checkpoint_after_ms,
              checkpoint_max_count=checkpoint_max_count,
              checkpoint_min_count=checkpoint_min_count,
              subscriber_max_count=subscriber_max_count,
              consumer_strategy=consumer_strategy,
              fail_if_exists=fail_if_exists,
              credentials=credentials,
          )
          return action
  
      async def connect_subscription(
          self, subscription: str, stream: str, conversation_id: uuid.UUID = None
      ):
          return subscriptions.Subscription(self._connection, stream, subscription)
  
      async def subscribe_to(
          self,
          stream: str,
          start_from: tg.Union[int, dts.Position] = dts.Position.HEAD,
          resolve_link_tos: bool = True,
          batch_size: int = 100,
      ):
          action = reader.Iterator(
              self._connection,
              link=dts.Link(
                  stream, start_from, dts.StreamDirection.Forward, count=batch_size
              ),
              resolve_link_tos=resolve_link_tos,
          )
          async for event in action.read(long_poll=60):
              yield event
  
      async def __aenter__(self) -> "Client":
          return self
  
      async def __aexit__(self, exc_type, exc, tb):
          await self._connection.close()
  
  
  def connect(base_url: str = "http://127.0.0.1:2113", loop=None):
      return Client(base_url=base_url, loop=loop)
  