Add top level API response model

smkent · Feb 20, 2023 · 840cba8 · 840cba8
1 parent 6489aab
commit 840cba8
Show file tree

Hide file tree

Showing 2 changed files with 46 additions and 25 deletions.
diff --git a/jmapc/api.py b/jmapc/api.py
@@ -0,0 +1,43 @@
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from typing import Any, Dict, List, Sequence, Tuple, Type
+
+from dataclasses_json import config
+
+from . import errors
+from .methods import (
+    CustomResponse,
+    InvocationResponseOrError,
+    Response,
+    ResponseOrError,
+)
+from .serializer import Model
+
+
+def decode_method_responses(
+    value: Sequence[Tuple[str, Dict[str, Any], str]],
+) -> List[InvocationResponseOrError]:
+    def _response_type(method_name: str) -> Type[ResponseOrError]:
+        if method_name == "error":
+            return errors.Error
+        return Response.response_types.get(method_name, CustomResponse)
+
+    return [
+        InvocationResponseOrError(
+            id=method_id,
+            response=_response_type(name).from_dict(response),
+        )
+        for name, response, method_id in value
+    ]
+
+
+@dataclass
+class APIResponse(Model):
+    method_responses: List[InvocationResponseOrError] = field(
+        metadata=config(
+            encoder=lambda value: None,
+            decoder=decode_method_responses,
+        ),
+    )
+    created_ids: List[str] = field(default_factory=list)
diff --git a/jmapc/client.py b/jmapc/client.py
@@ -25,10 +25,10 @@
 import sseclient
 
 from . import constants, errors
+from .api import APIResponse
 from .auth import BearerAuth
 from .logging import log
 from .methods import (
-    CustomResponse,
     Invocation,
     InvocationResponse,
     InvocationResponseOrError,
@@ -308,27 +308,5 @@ def _api_request(
         )
         r.raise_for_status()
         log.debug(f"Received JMAP response {r.text}")
-        return self._parse_method_responses(r.json())
-
-    def _parse_method_responses(
-        self, data: dict[str, Any]
-    ) -> Sequence[InvocationResponseOrError]:
-        method_responses = cast(
-            Sequence[Tuple[str, Dict[str, Any], str]],
-            data.get("methodResponses", []),
-        )
-
-        return [
-            InvocationResponseOrError(
-                id=method_id,
-                response=self._response_type(name).from_dict(response),
-            )
-            for name, response, method_id in method_responses
-        ]
-
-    def _response_type(self, method_name: str) -> Type[ResponseOrError]:
-        if method_name == "error":
-            return errors.Error
-        if method_name in Response.response_types:
-            return Response.response_types[method_name]
-        return CustomResponse
+        api_response = APIResponse.from_dict(r.json())
+        return api_response.method_responses