more wip

Author: Marenz

pyproject.toml

@@ -36,10 +36,10 @@ classifiers = [
 ]
 requires-python = ">= 3.11, < 4"
 dependencies = [
-  "typing-extensions >= 4.6.1, < 5",
+  "typing-extensions >= 4.13.0, < 5",
   "frequenz-api-dispatch == 1.0.0-rc2",
   "frequenz-client-base >= 0.8.0, < 0.12.0",
-  "frequenz-client-common >= 0.1.0, < 0.4.0",
+  "frequenz-client-common >= 0.3.2,  < 0.4.0",
   "grpcio >= 1.70.0, < 2",
   "python-dateutil >= 2.8.2, < 3.0",
 ]

src/frequenz/client/dispatch/_client.py

@@ -85,7 +85,8 @@ def __init__(
         )
         self._metadata = (("key", key),)
         self._streams: dict[
-            int, GrpcStreamBroadcaster[StreamMicrogridDispatchesResponse, DispatchEvent]
+            MicrogridId,
+            GrpcStreamBroadcaster[StreamMicrogridDispatchesResponse, DispatchEvent],
         ] = {}
         """A dictionary of streamers, keyed by microgrid_id."""
 
@@ -140,7 +141,7 @@ async def list(
             key="key",
             server_url="grpc://dispatch.url.goes.here.example.com"
         )
-        async for page in client.list(microgrid_id=1):
+        async for page in client.list(microgrid_id=MicrogridId(1)):
             for dispatch in page:
                 print(dispatch)
         ```
@@ -220,17 +221,6 @@ def stream(self, microgrid_id: MicrogridId) -> channels.Receiver[DispatchEvent]:
         dispatch events.
         An event is one of [CREATE, UPDATE, DELETE].
 
-        Example usage:
-
-        ```
-        client = DispatchApiClient(
-            key="key",
-            server_url="grpc://dispatch.url.goes.here.example.com"
-        )
-        async for message in client.stream(microgrid_id=1):
-            print(message.event, message.dispatch)
-        ```
-
         Args:
             microgrid_id: The microgrid_id to receive dispatches for.
 
@@ -361,7 +351,7 @@ async def update(
             ValueError: If updating `type` or `dry_run`.
         """
         msg = UpdateMicrogridDispatchRequest(
-            dispatch_id=dispatch_id, microgrid_id=microgrid_id
+            dispatch_id=int(dispatch_id), microgrid_id=int(microgrid_id)
         )
 
         for key, val in new_fields.items():

src/frequenz/client/dispatch/_internal_types.py

@@ -12,7 +12,9 @@
 from frequenz.api.dispatch.v1.dispatch_pb2 import (
     CreateMicrogridDispatchRequest as PBDispatchCreateRequest,
 )
-from frequenz.api.dispatch.v1.dispatch_pb2 import DispatchData
+from frequenz.api.dispatch.v1.dispatch_pb2 import (
+    DispatchData,
+)
 from google.protobuf.json_format import MessageToDict
 from google.protobuf.struct_pb2 import Struct
 from google.protobuf.timestamp_pb2 import Timestamp

src/frequenz/client/dispatch/test/_service.py

@@ -38,7 +38,7 @@
 from frequenz.client.common.microgrid import MicrogridId
 
 from .._internal_types import DispatchCreateRequest
-from ..types import Dispatch, DispatchEvent, Event
+from ..types import Dispatch, DispatchEvent, DispatchId, Event
 
 ALL_KEY = "all"
 """Key that has access to all resources in the FakeService."""
@@ -70,9 +70,20 @@ def __init__(self) -> None:
         self.dispatches: dict[MicrogridId, list[Dispatch]] = {}
         """List of dispatches per microgrid."""
 
-        self._last_id: int = 0
+        self._last_id: DispatchId = DispatchId(0)
         """Last used dispatch id."""
 
+    def refresh_last_id_for(self, microgrid_id: MicrogridId) -> None:
+        """Update last id to be the next highest number."""
+        dispatches = self.dispatches.get(microgrid_id, [])
+
+        if len(dispatches) == 0:
+            return
+
+        self._last_id = DispatchId(
+            max(int(self._last_id), max(int(dispatch.id) for dispatch in dispatches))
+        )
+
     def _check_access(self, metadata: grpc.aio.Metadata) -> None:
         """Check if the access key is valid.
 
@@ -170,7 +181,7 @@ async def StreamMicrogridDispatches(
 
         async for message in receiver:
             logging.debug("Received message: %s", message)
-            if message.microgrid_id == request.microgrid_id:
+            if message.microgrid_id == MicrogridId(request.microgrid_id):
                 response = StreamMicrogridDispatchesResponse(
                     event=message.event.event.value,
                     dispatch=message.event.dispatch.to_protobuf(),
@@ -225,7 +236,8 @@ async def CreateMicrogridDispatch(
     ) -> CreateMicrogridDispatchResponse:
         """Create a new dispatch."""
         self._check_access(metadata)
-        self._last_id += 1
+        microgrid_id = MicrogridId(request.microgrid_id)
+        self._last_id = DispatchId(int(self._last_id) + 1)
 
         new_dispatch = _dispatch_from_request(
             DispatchCreateRequest.from_protobuf(request),
@@ -235,11 +247,11 @@ async def CreateMicrogridDispatch(
         )
 
         # implicitly create the list if it doesn't exist
-        self.dispatches.setdefault(request.microgrid_id, []).append(new_dispatch)
+        self.dispatches.setdefault(microgrid_id, []).append(new_dispatch)
 
         await self._stream_sender.send(
             self.StreamEvent(
-                request.microgrid_id,
+                microgrid_id,
                 DispatchEvent(dispatch=new_dispatch, event=Event.CREATED),
             )
         )
@@ -254,9 +266,15 @@ async def UpdateMicrogridDispatch(
     ) -> UpdateMicrogridDispatchResponse:
         """Update a dispatch."""
         self._check_access(metadata)
-        grid_dispatches = self.dispatches[request.microgrid_id]
+
+        microgrid_id = MicrogridId(request.microgrid_id)
+        grid_dispatches = self.dispatches.get(microgrid_id, [])
         index = next(
-            (i for i, d in enumerate(grid_dispatches) if d.id == request.dispatch_id),
+            (
+                i
+                for i, d in enumerate(grid_dispatches)
+                if d.id == DispatchId(request.dispatch_id)
+            ),
             None,
         )
 
@@ -313,9 +331,9 @@ async def UpdateMicrogridDispatch(
                             | "bymonthdays"
                             | "bymonths"
                         ):
-                            getattr(pb_dispatch.data.recurrence, split_path[1])[
-                                :
-                            ] = getattr(request.update.recurrence, split_path[1])[:]
+                            getattr(pb_dispatch.data.recurrence, split_path[1])[:] = (
+                                getattr(request.update.recurrence, split_path[1])[:]
+                            )
 
         dispatch = Dispatch.from_protobuf(pb_dispatch)
         dispatch = replace(
@@ -327,7 +345,7 @@ async def UpdateMicrogridDispatch(
 
         await self._stream_sender.send(
             self.StreamEvent(
-                request.microgrid_id,
+                microgrid_id,
                 DispatchEvent(dispatch=dispatch, event=Event.UPDATED),
             )
         )
@@ -342,9 +360,11 @@ async def GetMicrogridDispatch(
     ) -> GetMicrogridDispatchResponse:
         """Get a single dispatch."""
         self._check_access(metadata)
-        grid_dispatches = self.dispatches.get(request.microgrid_id, [])
+        microgrid_id = MicrogridId(request.microgrid_id)
+        grid_dispatches = self.dispatches.get(microgrid_id, [])
         dispatch = next(
-            (d for d in grid_dispatches if d.id == request.dispatch_id), None
+            (d for d in grid_dispatches if d.id == DispatchId(request.dispatch_id)),
+            None,
         )
 
         if dispatch is None:
@@ -365,10 +385,12 @@ async def DeleteMicrogridDispatch(
     ) -> Empty:
         """Delete a given dispatch."""
         self._check_access(metadata)
-        grid_dispatches = self.dispatches.get(request.microgrid_id, [])
+        microgrid_id = MicrogridId(request.microgrid_id)
+        grid_dispatches = self.dispatches.get(microgrid_id, [])
 
         dispatch_to_delete = next(
-            (d for d in grid_dispatches if d.id == request.dispatch_id), None
+            (d for d in grid_dispatches if d.id == DispatchId(request.dispatch_id)),
+            None,
         )
 
         if dispatch_to_delete is None:
@@ -383,7 +405,7 @@ async def DeleteMicrogridDispatch(
 
         await self._stream_sender.send(
             self.StreamEvent(
-                request.microgrid_id,
+                microgrid_id,
                 DispatchEvent(
                     dispatch=dispatch_to_delete,
                     event=Event.DELETED,
@@ -398,7 +420,7 @@ async def DeleteMicrogridDispatch(
 
 def _dispatch_from_request(
     _request: DispatchCreateRequest,
-    _id: int,
+    _id: DispatchId,
     create_time: datetime,
     update_time: datetime,
 ) -> Dispatch:

src/frequenz/client/dispatch/test/client.py

@@ -55,16 +55,7 @@ def set_dispatches(self, microgrid_id: MicrogridId, value: list[Dispatch]) -> No
             value: The list of dispatches to set.
         """
         self._service.dispatches[microgrid_id] = value
-
-        if len(value) == 0:
-            return
-
-        # Max between last id and the max id in the list
-        # pylint: disable=protected-access
-        self._service._last_id = max(
-            self._service._last_id, max(dispatch.id for dispatch in value)
-        )
-        # pylint: enable=protected-access
+        self._service.refresh_last_id_for(microgrid_id)
 
     @property
     def _service(self) -> FakeService: