Update the client-base dependency to v0.8.0

We now use the new suggested way to get the `stub`, so it has proper
async type hints.

Author: llucax

pyproject.toml

@@ -38,7 +38,8 @@ requires-python = ">= 3.11, < 4"
 dependencies = [
   "frequenz-api-microgrid >= 0.15.3, < 0.16.0",
   "frequenz-channels >= 1.0.0-rc1, < 2.0.0",
-  "frequenz-client-base >= 0.6.0, < 0.7",
+  #"frequenz-client-base >= 0.8.0, < 0.9",
+  "frequenz-client-base @ git+https://github.com/llucax/frequenz-client-base-python.git@stub-back#egg=frequenz-client-base",
   "grpcio >= 1.54.2, < 2",
   "protobuf >= 4.21.6, < 6",
   "timezonefinder >= 6.2.0, < 7",

src/frequenz/client/microgrid/_client.py

@@ -3,6 +3,8 @@
 
 """Client for requests to the Microgrid API."""
 
+from __future__ import annotations
+
 import asyncio
 import logging
 from collections.abc import Callable, Iterable, Set
@@ -31,7 +33,7 @@
 )
 from ._connection import Connection
 from ._constants import RECEIVER_MAX_SIZE
-from ._exception import ApiClientError
+from ._exception import ApiClientError, ClientNotConnected
 from ._metadata import Location, Metadata
 
 DEFAULT_GRPC_CALL_TIMEOUT = 60.0
@@ -89,10 +91,20 @@ def __init__(
             connect=connect,
             channel_defaults=channel_defaults,
         )
-        self._async_stub: microgrid_pb2_grpc.MicrogridAsyncStub = self.stub  # type: ignore
         self._broadcasters: dict[int, streaming.GrpcStreamBroadcaster[Any, Any]] = {}
         self._retry_strategy = retry_strategy
 
+    @property
+    def stub(self) -> microgrid_pb2_grpc.MicrogridAsyncStub:
+        """The gRPC stub for the API."""
+        if self.channel is None or self._stub is None:
+            raise ClientNotConnected(server_url=self.server_url, operation="stub")
+        # This type: ignore is needed because we need to cast the sync stub to
+        # the async stub, but we can't use cast because the async stub doesn't
+        # actually exists to the eyes of the interpreter, it only exists for the
+        # type-checker, so it can only be used for type hints.
+        return self._stub  # type: ignore
+
     async def components(  # noqa: DOC502 (raises ApiClientError indirectly)
         self,
     ) -> Iterable[Component]:
@@ -108,7 +120,7 @@ async def components(  # noqa: DOC502 (raises ApiClientError indirectly)
         """
         component_list = await client.call_stub_method(
             self,
-            lambda: self._async_stub.ListComponents(
+            lambda: self.stub.ListComponents(
                 microgrid_pb2.ComponentFilter(),
                 timeout=int(DEFAULT_GRPC_CALL_TIMEOUT),
             ),
@@ -145,7 +157,7 @@ async def metadata(self) -> Metadata:
         try:
             microgrid_metadata = await client.call_stub_method(
                 self,
-                lambda: self._async_stub.GetMicrogridMetadata(
+                lambda: self.stub.GetMicrogridMetadata(
                     Empty(),
                     timeout=int(DEFAULT_GRPC_CALL_TIMEOUT),
                 ),
@@ -192,7 +204,7 @@ async def connections(  # noqa: DOC502 (raises ApiClientError indirectly)
             self.components(),
             client.call_stub_method(
                 self,
-                lambda: self._async_stub.ListConnections(
+                lambda: self.stub.ListConnections(
                     connection_filter,
                     timeout=int(DEFAULT_GRPC_CALL_TIMEOUT),
                 ),
@@ -249,12 +261,15 @@ async def _new_component_data_receiver(
             broadcaster = streaming.GrpcStreamBroadcaster(
                 f"raw-component-data-{component_id}",
                 lambda: aiter(
-                    self._async_stub.StreamComponentData(
+                    self.stub.StreamComponentData(
                         microgrid_pb2.ComponentIdParam(id=component_id)
                     )
                 ),
                 transform,
                 retry_strategy=self._retry_strategy,
+                # We don't expect any data stream to end, so if it is exhausted for any
+                # reason we want to keep retrying
+                retry_on_exhausted_stream=True,
             )
             self._broadcasters[component_id] = broadcaster
         return broadcaster.new_receiver(maxsize=maxsize)
@@ -408,7 +423,7 @@ async def set_power(  # noqa: DOC502 (raises ApiClientError indirectly)
         """
         await client.call_stub_method(
             self,
-            lambda: self._async_stub.SetPowerActive(
+            lambda: self.stub.SetPowerActive(
                 microgrid_pb2.SetPowerActiveParam(
                     component_id=component_id, power=power_w
                 ),
@@ -447,7 +462,7 @@ async def set_bounds(  # noqa: DOC503 (raises ApiClientError indirectly)
         )
         await client.call_stub_method(
             self,
-            lambda: self._async_stub.AddInclusionBounds(
+            lambda: self.stub.AddInclusionBounds(
                 microgrid_pb2.SetBoundsParam(
                     component_id=component_id,
                     target_metric=target_metric,

tests/test_client.py

@@ -47,7 +47,6 @@ def __init__(self, *, retry_strategy: retry.Strategy | None = None) -> None:
         super().__init__("grpc://mock_host:1234", retry_strategy=retry_strategy)
         self.mock_stub = mock_stub
         self._stub = mock_stub  # pylint: disable=protected-access
-        self._async_stub = mock_stub  # pylint: disable=protected-access
 
 
 async def test_components() -> None: