add PostsResponse; harmonize get_messages and watch_messages parameters; refactor node.py; increase test coverage

Author: MHHukiewitz

src/aleph/sdk/base.py

@@ -26,6 +26,7 @@
 from aleph_message.models.execution.program import Encoding
 from aleph_message.status import MessageStatus
 
+from aleph.sdk.models import PostsResponse
 from aleph.sdk.types import GenericMessage, StorageEnum
 
 DEFAULT_PAGE_SIZE = 200
@@ -78,7 +79,9 @@ async def get_posts(
         chains: Optional[Iterable[str]] = None,
         start_date: Optional[Union[datetime, float]] = None,
         end_date: Optional[Union[datetime, float]] = None,
-    ) -> Dict[str, Dict]:
+        ignore_invalid_messages: bool = True,
+        invalid_messages_log_level: int = logging.NOTSET,
+    ) -> PostsResponse:
         """
         Fetch a list of posts from the network.
 
@@ -93,6 +96,8 @@ async def get_posts(
         :param chains: Chains of the posts to fetch (Default: all chains)
         :param start_date: Earliest date to fetch messages from
         :param end_date: Latest date to fetch messages from
+        :param ignore_invalid_messages: Ignore invalid messages (Default: True)
+        :param invalid_messages_log_level: Log level to use for invalid messages (Default: logging.NOTSET)
         """
         pass
 
@@ -138,9 +143,9 @@ async def get_posts_iterator(
                 start_date=start_date,
                 end_date=end_date,
             )
-            total_items = resp["pagination_total"]
+            total_items = resp.pagination_total
             page += 1
-            for post in resp["posts"]:
+            for post in resp.posts:
                 yield post
 
     @abstractmethod
@@ -183,7 +188,7 @@ async def get_messages(
         :param page: Page to fetch, begins at 1 (Default: 1)
         :param message_type: Filter by message type, can be "AGGREGATE", "POST", "PROGRAM", "VM", "STORE" or "FORGET"
         :param content_types: Filter by content type
-        :param content_keys: Filter by content key
+        :param content_keys: Filter by aggregate key
         :param refs: If set, only fetch posts that reference these hashes (in the "refs" field)
         :param addresses: Addresses of the posts to fetch (Default: all addresses)
         :param tags: Tags of the posts to fetch (Default: all tags)
@@ -192,7 +197,7 @@ async def get_messages(
         :param chains: Filter by sender address chain
         :param start_date: Earliest date to fetch messages from
         :param end_date: Latest date to fetch messages from
-        :param ignore_invalid_messages: Ignore invalid messages (Default: False)
+        :param ignore_invalid_messages: Ignore invalid messages (Default: True)
         :param invalid_messages_log_level: Log level to use for invalid messages (Default: logging.NOTSET)
         """
         pass
@@ -271,6 +276,7 @@ def watch_messages(
         self,
         message_type: Optional[MessageType] = None,
         content_types: Optional[Iterable[str]] = None,
+        content_keys: Optional[Iterable[str]] = None,
         refs: Optional[Iterable[str]] = None,
         addresses: Optional[Iterable[str]] = None,
         tags: Optional[Iterable[str]] = None,
@@ -285,6 +291,7 @@ def watch_messages(
 
         :param message_type: Type of message to watch
         :param content_types: Content types to watch
+        :param content_keys: Filter by aggregate key
         :param refs: References to watch
         :param addresses: Addresses to watch
         :param tags: Tags to watch

src/aleph/sdk/client.py

@@ -59,7 +59,7 @@
     MessageNotFoundError,
     MultipleMessagesError,
 )
-from .models import MessagesResponse
+from .models import MessagesResponse, PostsResponse
 from .utils import check_unix_socket_valid, get_message_type_value
 
 logger = logging.getLogger(__name__)
@@ -215,7 +215,7 @@ def get_posts(
         chains: Optional[Iterable[str]] = None,
         start_date: Optional[Union[datetime, float]] = None,
         end_date: Optional[Union[datetime, float]] = None,
-    ) -> Dict[str, Dict]:
+    ) -> PostsResponse:
         return self._wrap(
             self.async_session.get_posts,
             pagination=pagination,
@@ -575,7 +575,18 @@ async def get_posts(
         chains: Optional[Iterable[str]] = None,
         start_date: Optional[Union[datetime, float]] = None,
         end_date: Optional[Union[datetime, float]] = None,
-    ) -> Dict[str, Dict]:
+        ignore_invalid_messages: bool = True,
+        invalid_messages_log_level: int = logging.NOTSET,
+    ) -> PostsResponse:
+        ignore_invalid_messages = (
+            True if ignore_invalid_messages is None else ignore_invalid_messages
+        )
+        invalid_messages_log_level = (
+            logging.NOTSET
+            if invalid_messages_log_level is None
+            else invalid_messages_log_level
+        )
+
         params: Dict[str, Any] = dict(pagination=pagination, page=page)
 
         if types is not None:
@@ -604,7 +615,36 @@ async def get_posts(
 
         async with self.http_session.get("/api/v0/posts.json", params=params) as resp:
             resp.raise_for_status()
-            return await resp.json()
+            response_json = await resp.json()
+            posts_raw = response_json["posts"]
+
+            # All posts may not be valid according to the latest specification in
+            # aleph-message. This allows the user to specify how errors should be handled.
+            posts: List[AlephMessage] = []
+            for post_raw in posts_raw:
+                try:
+                    message = parse_message(post_raw)
+                    posts.append(message)
+                except KeyError as e:
+                    if not ignore_invalid_messages:
+                        raise e
+                    logger.log(
+                        level=invalid_messages_log_level,
+                        msg=f"KeyError: Field '{e.args[0]}' not found",
+                    )
+                except ValidationError as e:
+                    if not ignore_invalid_messages:
+                        raise e
+                    if invalid_messages_log_level:
+                        logger.log(level=invalid_messages_log_level, msg=e)
+
+            return PostsResponse(
+                posts=posts,
+                pagination_page=response_json["pagination_page"],
+                pagination_total=response_json["pagination_total"],
+                pagination_per_page=response_json["pagination_per_page"],
+                pagination_item=response_json["pagination_item"],
+            )
 
     async def download_file_to_buffer(
         self,
@@ -807,6 +847,7 @@ async def watch_messages(
         self,
         message_type: Optional[MessageType] = None,
         content_types: Optional[Iterable[str]] = None,
+        content_keys: Optional[Iterable[str]] = None,
         refs: Optional[Iterable[str]] = None,
         addresses: Optional[Iterable[str]] = None,
         tags: Optional[Iterable[str]] = None,
@@ -822,6 +863,8 @@ async def watch_messages(
             params["msgType"] = message_type.value
         if content_types is not None:
             params["contentTypes"] = ",".join(content_types)
+        if content_keys is not None:
+            params["contentKeys"] = ",".join(content_keys)
         if refs is not None:
             params["refs"] = ",".join(refs)
         if addresses is not None:

src/aleph/sdk/models.py

@@ -1,14 +1,25 @@
 from typing import List
 
-from aleph_message.models import AlephMessage
+from aleph_message.models import AlephMessage, PostMessage
 from pydantic import BaseModel
 
 
-class MessagesResponse(BaseModel):
-    """Response from an Aleph node API on the path /api/v0/messages.json"""
-
-    messages: List[AlephMessage]
+class PaginationResponse(BaseModel):
     pagination_page: int
     pagination_total: int
     pagination_per_page: int
     pagination_item: str
+
+
+class MessagesResponse(PaginationResponse):
+    """Response from an Aleph node API on the path /api/v0/messages.json"""
+
+    messages: List[AlephMessage]
+    pagination_item = "messages"
+
+
+class PostsResponse(PaginationResponse):
+    """Response from an Aleph node API on the path /api/v0/posts.json"""
+
+    posts: List[PostMessage]
+    pagination_item = "posts"