fetch report from kes

examples/example.py

@@ -65,3 +65,7 @@
     del parent_table[0]
 parent_row = CategoryParentAssetRow(relationship=ref)
 parent_table.append_row(parent_row)
+
+#download report and save it to a file
+with open("report.docx", "wb") as binary_file:
+    binary_file.write(activity.download_report())

setup.cfg

@@ -1,6 +1,6 @@
 [metadata]
 name = kes-RHDHV
-version = 1.dev9
+version = 1.dev10
 author = RoyalHaskoningDHV
 author_email = [email protected]
 description = KES python client

src/kes/client.py

@@ -129,4 +129,4 @@ def open_activity_by_id(self, activity_id: UUID = os.getenv('ACTIVITY_ID', ''))
         Returns:
             An instance representing the requested Kes Activity.
         """
-        return Activity(self._table_stub, activity_id, "")
+        return Activity(self._table_stub, activity_id, "", self._project_stub)

src/kes/project.py

@@ -5,10 +5,10 @@
 
 
 from functools import cached_property
-from typing import List
+from typing import List, Optional, ByteString
 from uuid import UUID
 
-from kes.proto.project_pb2 import ReadActivitiesReply, ReadActivitiesRequest
+from kes.proto.project_pb2 import ReadActivitiesReply, ReadActivitiesRequest, DownloadReportRequest
 from kes.proto.project_pb2_grpc import ProjectDetailStub
 from kes.proto.table_pb2_grpc import TableStub
 from kes.table import RowType, Table, TableDef
@@ -24,11 +24,13 @@ class Activity:
     _id: UUID
     _description: str
     _stub: TableStub
+    _project_stub: ProjectDetailStub
 
-    def __init__(self, stub: TableStub, id: UUID, description: str):
+    def __init__(self, stub: TableStub, id: UUID, description: str, project_stub: ProjectDetailStub):
         self._id = id
         self._stub = stub
         self._description = description
+        self._project_stub = project_stub
 
     def build_table(self, table_def: TableDef[RowType]) -> Table[RowType]:
         """ Create a table for this activity.
@@ -48,6 +50,21 @@ def build_table(self, table_def: TableDef[RowType]) -> Table[RowType]:
             table_def.property_map
         )
 
+    def download_report(self) -> Optional[ByteString]:
+        """ Download Report for the activity.
+
+        Returns:
+            Activity Report.
+        """
+        reportData = bytearray()
+        request = DownloadReportRequest(activityId=str(self._id))
+        streamingReply = self._project_stub.downloadReport(request)
+
+        for reply in streamingReply:
+            reportData += reply.chunk
+
+        return reportData
+
     @property
     def id(self) -> UUID:
         """uuid: id of the activity."""