Stable video stats API: GetCallReport

Author: su225

getstream/models/__init__.py

@@ -925,6 +925,29 @@ class CallRecording(DataClassJsonMixin):
     url: str = dc_field(metadata=dc_config(field_name="url"))
 
 
+@dataclass
+class CallReportResponse(DataClassJsonMixin):
+    score: float = dc_field(metadata=dc_config(field_name="score"))
+    ended_at: Optional[datetime] = dc_field(
+        default=None,
+        metadata=dc_config(
+            field_name="ended_at",
+            encoder=encode_datetime,
+            decoder=datetime_from_unix_ns,
+            mm_field=fields.DateTime(format="iso"),
+        ),
+    )
+    started_at: Optional[datetime] = dc_field(
+        default=None,
+        metadata=dc_config(
+            field_name="started_at",
+            encoder=encode_datetime,
+            decoder=datetime_from_unix_ns,
+            mm_field=fields.DateTime(format="iso"),
+        ),
+    )
+
+
 @dataclass
 class CallRequest(DataClassJsonMixin):
     created_by_id: Optional[str] = dc_field(
@@ -4257,6 +4280,13 @@ class GetCallResponse(DataClassJsonMixin):
     call: "CallResponse" = dc_field(metadata=dc_config(field_name="call"))
 
 
+@dataclass
+class GetCallReportResponse(DataClassJsonMixin):
+    duration: str = dc_field(metadata=dc_config(field_name="duration"))
+    session_id: str = dc_field(metadata=dc_config(field_name="session_id"))
+    report: "ReportResponse" = dc_field(metadata=dc_config(field_name="report"))
+
+
 @dataclass
 class GetCallStatsResponse(DataClassJsonMixin):
     call_duration_seconds: int = dc_field(
@@ -6408,6 +6438,12 @@ class PaginationParams(DataClassJsonMixin):
     )
 
 
+@dataclass
+class ParticipantReportResponse(DataClassJsonMixin):
+    sum: int = dc_field(metadata=dc_config(field_name="sum"))
+    unique: int = dc_field(metadata=dc_config(field_name="unique"))
+
+
 @dataclass
 class PendingMessageResponse(DataClassJsonMixin):
     channel: "Optional[ChannelResponse]" = dc_field(
@@ -8025,6 +8061,17 @@ class ReportByHistogramBucket(DataClassJsonMixin):
     )
 
 
+@dataclass
+class ReportResponse(DataClassJsonMixin):
+    call: "CallReportResponse" = dc_field(metadata=dc_config(field_name="call"))
+    participants: "ParticipantReportResponse" = dc_field(
+        metadata=dc_config(field_name="participants")
+    )
+    user_ratings: "UserRatingReportResponse" = dc_field(
+        metadata=dc_config(field_name="user_ratings")
+    )
+
+
 @dataclass
 class Response(DataClassJsonMixin):
     duration: str = dc_field(metadata=dc_config(field_name="duration"))
@@ -10482,6 +10529,12 @@ class UserMuteResponse(DataClassJsonMixin):
     )
 
 
+@dataclass
+class UserRatingReportResponse(DataClassJsonMixin):
+    average: float = dc_field(metadata=dc_config(field_name="average"))
+    count: int = dc_field(metadata=dc_config(field_name="count"))
+
+
 @dataclass
 class UserRequest(DataClassJsonMixin):
     id: str = dc_field(metadata=dc_config(field_name="id"))

getstream/video/rest_client.py

@@ -751,3 +751,19 @@ def query_aggregate_call_stats(
         return self.post(
             "/api/v2/video/stats", QueryAggregateCallStatsResponse, json=json
         )
+
+    def get_call_report(
+        self, type: str, id: str, session_id: Optional[str] = None
+    ) -> StreamResponse[GetCallReportResponse]:
+        query_params = build_query_param(session_id=session_id)
+        path_params = {
+            "type": type,
+            "id": id,
+        }
+
+        return self.get(
+            "/api/v2/video/call/{type}/{id}/report",
+            GetCallReportResponse,
+            query_params=query_params,
+            path_params=path_params,
+        )

tests/test_video_integration.py

@@ -355,6 +355,16 @@ def test_delete_not_existing_transcription(self):
         with pytest.raises(StreamAPIException):
             self.call.delete_transcription("random_session", "random_filename")
 
+    def test_get_call_report_for_latest_session(self):
+        with pytest.raises(StreamAPIException):
+            self.client.video.get_call_report(self.call.call_type, self.call.id)
+
+    def test_get_call_report_for_specified_session(self):
+        with pytest.raises(StreamAPIException):
+            self.client.video.get_call_report(
+                self.call.call_type, self.call.id, session_id="non_existent"
+            )
+
 
 class TestDeleteCall:
     def test_soft_delete(self, call: Call):