Use extended_json_encoder as default in ClientSession (#84)

* Use `extended_json_encoder` when dumping message contents everywhere

* Add `extended_json_encoder` as default to `ClientSession`

* Add additional tests for `extended_json_encoder` to prove timezone compatibility

* Add docstring to `extended_json_encoder`

Author: MHHukiewitz

src/aleph/sdk/client/authenticated_http.py

@@ -1,4 +1,3 @@
-import datetime
 import hashlib
 import json
 import logging
@@ -34,11 +33,11 @@
 from aleph_message.models.execution.program import CodeContent, FunctionRuntime
 from aleph_message.models.execution.volume import MachineVolume, ParentVolume
 from aleph_message.status import MessageStatus
-from pydantic.json import pydantic_encoder
 
 from ..conf import settings
 from ..exceptions import BroadcastError, InvalidMessageError
 from ..types import Account, StorageEnum
+from ..utils import extended_json_encoder
 from .abstract import AuthenticatedAlephClient
 from .http import AlephHttpClient
 
@@ -51,16 +50,6 @@
     magic = None  # type:ignore
 
 
-def extended_json_encoder(obj: Any) -> str:
-    if (
-        isinstance(obj, datetime.datetime)
-        or isinstance(obj, datetime.date)
-        or isinstance(obj, datetime.time)
-    ):
-        return obj.isoformat()  # or any other format you prefer
-    return pydantic_encoder(obj)
-
-
 class AuthenticatedAlephHttpClient(AlephHttpClient, AuthenticatedAlephClient):
     account: Account
 
@@ -181,12 +170,20 @@ async def _handle_broadcast_error(response: aiohttp.ClientResponse) -> NoReturn:
             logger.error(error_msg)
             raise BroadcastError(error_msg)
         elif response.status == 422:
-            errors = await response.json()
-            logger.error(
-                "The message could not be processed because of the following errors: %s",
-                errors,
-            )
-            raise InvalidMessageError(errors)
+            try:
+                errors = await response.json()
+                logger.error(
+                    "The message could not be processed because of the following errors: %s",
+                    errors,
+                )
+                raise InvalidMessageError(errors)
+            except (json.JSONDecodeError, aiohttp.client_exceptions.ContentTypeError):
+                error = await response.text()
+                logger.error(
+                    "The message could not be processed because of the following errors: %s",
+                    error,
+                )
+                raise InvalidMessageError(error)
         else:
             error_msg = (
                 f"Unexpected HTTP response ({response.status}: {await response.text()})"
@@ -212,12 +209,11 @@ async def _broadcast_deprecated(self, message_dict: Mapping[str, Any]) -> None:
 
         url = "/api/v0/ipfs/pubsub/pub"
         logger.debug(f"Posting message on {url}")
-
         async with self.http_session.post(
             url,
             json={
                 "topic": "ALEPH-TEST",
-                "data": json.dumps(message_dict, default=extended_json_encoder),
+                "data": message_dict,
             },
         ) as response:
             await self._handle_broadcast_deprecated_response(response)
@@ -257,10 +253,12 @@ async def _broadcast(
         logger.debug(f"Posting message on {url}")
 
         message_dict = message.dict(include=self.BROADCAST_MESSAGE_FIELDS)
-
         async with self.http_session.post(
             url,
-            json={"sync": sync, "message": message_dict},
+            json={
+                "sync": sync,
+                "message": message_dict,
+            },
         ) as response:
             # The endpoint may be unavailable on this node, try the deprecated version.
             if response.status in (404, 405):

src/aleph/sdk/client/http.py

@@ -17,6 +17,7 @@
     Writable,
     check_unix_socket_valid,
     copy_async_readable_to_buffer,
+    extended_json_encoder,
     get_message_type_value,
 )
 from .abstract import AlephClient
@@ -53,12 +54,18 @@ def __init__(
         # ClientSession timeout defaults to a private sentinel object and may not be None.
         self.http_session = (
             aiohttp.ClientSession(
-                base_url=self.api_server, connector=connector, timeout=timeout
+                base_url=self.api_server,
+                connector=connector,
+                timeout=timeout,
+                json_serialize=extended_json_encoder,
             )
             if timeout
             else aiohttp.ClientSession(
                 base_url=self.api_server,
                 connector=connector,
+                json_serialize=lambda obj: json.dumps(
+                    obj, default=extended_json_encoder
+                ),
             )
         )
 

src/aleph/sdk/utils.py

@@ -1,15 +1,16 @@
 import errno
 import logging
 import os
-from datetime import datetime
+from datetime import date, datetime, time
 from enum import Enum
 from pathlib import Path
 from shutil import make_archive
-from typing import Iterable, Optional, Protocol, Tuple, Type, TypeVar, Union
+from typing import Any, Iterable, Optional, Protocol, Tuple, Type, TypeVar, Union
 from zipfile import BadZipFile, ZipFile
 
 from aleph_message.models import MessageType
 from aleph_message.models.execution.program import Encoding
+from pydantic.json import pydantic_encoder
 
 from aleph.sdk.conf import settings
 from aleph.sdk.types import GenericMessage
@@ -135,3 +136,17 @@ def _date_field_to_timestamp(date: Optional[Union[datetime, float]]) -> Optional
         return str(date.timestamp())
     else:
         raise TypeError(f"Invalid type: `{type(date)}`")
+
+
+def extended_json_encoder(obj: Any) -> Any:
+    """
+    Extended JSON encoder for dumping objects that contain pydantic models and datetime objects.
+    """
+    if isinstance(obj, datetime):
+        return obj.timestamp()
+    elif isinstance(obj, date):
+        return obj.toordinal()
+    elif isinstance(obj, time):
+        return obj.hour * 3600 + obj.minute * 60 + obj.second + obj.microsecond / 1e6
+    else:
+        return pydantic_encoder(obj)

tests/unit/test_asynchronous.py

@@ -1,17 +1,14 @@
-import datetime
 from unittest.mock import AsyncMock
 
 import pytest as pytest
 from aleph_message.models import (
     AggregateMessage,
     ForgetMessage,
     InstanceMessage,
-    MessageType,
     PostMessage,
     ProgramMessage,
     StoreMessage,
 )
-from aleph_message.models.execution.environment import MachineResources
 from aleph_message.status import MessageStatus
 
 from aleph.sdk.types import StorageEnum
@@ -124,63 +121,3 @@ async def test_forget(mock_session_with_post_success):
 
     assert mock_session_with_post_success.http_session.post.called_once
     assert isinstance(forget_message, ForgetMessage)
-
-
-@pytest.mark.asyncio
-@pytest.mark.parametrize(
-    "message_type, content",
-    [
-        (
-            MessageType.aggregate,
-            {
-                "content": {"Hello": datetime.datetime.now()},
-                "key": "test",
-                "address": "0x1",
-                "time": 1.0,
-            },
-        ),
-        (
-            MessageType.aggregate,
-            {
-                "content": {"Hello": datetime.date.today()},
-                "key": "test",
-                "address": "0x1",
-                "time": 1.0,
-            },
-        ),
-        (
-            MessageType.aggregate,
-            {
-                "content": {"Hello": datetime.time()},
-                "key": "test",
-                "address": "0x1",
-                "time": 1.0,
-            },
-        ),
-        (
-            MessageType.aggregate,
-            {
-                "content": {
-                    "Hello": MachineResources(
-                        vcpus=1,
-                        memory=1024,
-                        seconds=1,
-                    )
-                },
-                "key": "test",
-                "address": "0x1",
-                "time": 1.0,
-            },
-        ),
-    ],
-)
-async def test_prepare_aleph_message(
-    mock_session_with_post_success, message_type, content
-):
-    # Call the function under test
-    async with mock_session_with_post_success as session:
-        await session._prepare_aleph_message(
-            message_type=message_type,
-            content=content,
-            channel="TEST",
-        )

tests/unit/test_utils.py

@@ -1,3 +1,6 @@
+import datetime
+
+import pytest as pytest
 from aleph_message.models import (
     AggregateMessage,
     Chain,
@@ -8,6 +11,7 @@
     ProgramMessage,
     StoreMessage,
 )
+from aleph_message.models.execution.environment import MachineResources
 
 from aleph.sdk.utils import enum_as_str, get_message_type_value
 
@@ -26,3 +30,101 @@ def test_enum_as_str():
     assert enum_as_str(Chain.ETH) == "ETH"
     assert enum_as_str(ItemType("inline")) == "inline"
     assert enum_as_str(ItemType.inline) == "inline"
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    "message_type, content",
+    [
+        (
+            MessageType.aggregate,
+            {
+                "content": {"Hello": datetime.datetime.now()},
+                "key": "test",
+                "address": "0x1",
+                "time": 1.0,
+            },
+        ),
+        (
+            MessageType.aggregate,
+            {
+                "content": {"Hello": datetime.date.today()},
+                "key": "test",
+                "address": "0x1",
+                "time": 1.0,
+            },
+        ),
+        (
+            MessageType.aggregate,
+            {
+                "content": {"Hello": datetime.time()},
+                "key": "test",
+                "address": "0x1",
+                "time": 1.0,
+            },
+        ),
+        (
+            MessageType.aggregate,
+            {
+                "content": {"Hello": datetime.timedelta()},
+                "key": "test",
+                "address": "0x1",
+                "time": 1.0,
+            },
+        ),
+        (
+            MessageType.aggregate,
+            {
+                "content": {"Hello": datetime.datetime.now().astimezone()},
+                "key": "test",
+                "address": "0x1",
+                "time": 1.0,
+            },
+        ),
+        (
+            MessageType.aggregate,
+            {
+                "content": {"Hello": datetime.datetime.now().astimezone().isoformat()},
+                "key": "test",
+                "address": "0x1",
+                "time": 1.0,
+            },
+        ),
+        (
+            MessageType.aggregate,
+            {
+                "content": {"Hello": datetime.datetime.now().astimezone().timestamp()},
+                "key": "test",
+                "address": "0x1",
+                "time": 1.0,
+            },
+        ),
+        (
+            MessageType.aggregate,
+            {
+                "content": {
+                    "Hello": MachineResources(
+                        vcpus=1,
+                        memory=1024,
+                        seconds=1,
+                    )
+                },
+                "key": "test",
+                "address": "0x1",
+                "time": 1.0,
+            },
+        ),
+    ],
+)
+async def test_prepare_aleph_message(
+    mock_session_with_post_success, message_type, content
+):
+    # Call the function under test
+    async with mock_session_with_post_success as session:
+        message = await session._prepare_aleph_message(
+            message_type=message_type,
+            content=content,
+            channel="TEST",
+        )
+
+    assert message.content.dict() == content