refactor interface; implement node methods; add download_file() metho…

…d to cache as

src/aleph/sdk/cache.py

@@ -152,6 +152,9 @@ def __getitem__(self, item_hash) -> Optional[AlephMessage]:
     def __setitem__(self, item_hash, message: AlephMessage):
         MessageModel.insert(**message_to_model(message)).on_conflict_replace().execute()
 
+    def __delitem__(self, item_hash):
+        MessageModel.delete().where(MessageModel.item_hash == item_hash).execute()
+
     def __contains__(self, item_hash):
         return MessageModel.select().where(MessageModel.item_hash == item_hash).exists()
 

src/aleph/sdk/client.py

@@ -898,6 +898,11 @@ async def __aenter__(self) -> "AuthenticatedAlephClient":
         return self
 
     async def ipfs_push(self, content: Mapping) -> str:
+        """
+        Push arbitrary content as JSON to the IPFS service.
+
+        :param content: The dict-like content to upload
+        """
         url = "/api/v0/ipfs/add_json"
         logger.debug(f"Pushing to IPFS on {url}")
 
@@ -906,6 +911,11 @@ async def ipfs_push(self, content: Mapping) -> str:
             return (await resp.json()).get("hash")
 
     async def storage_push(self, content: Mapping) -> str:
+        """
+        Push arbitrary content as JSON to the storage service.
+
+        :param content: The dict-like content to upload
+        """
         url = "/api/v0/storage/add_json"
         logger.debug(f"Pushing to storage on {url}")
 
@@ -914,6 +924,11 @@ async def storage_push(self, content: Mapping) -> str:
             return (await resp.json()).get("hash")
 
     async def ipfs_push_file(self, file_content: Union[str, bytes]) -> str:
+        """
+        Push a file to the IPFS service.
+
+        :param file_content: The file content to upload
+        """
         data = aiohttp.FormData()
         data.add_field("file", file_content)
 
@@ -925,6 +940,9 @@ async def ipfs_push_file(self, file_content: Union[str, bytes]) -> str:
             return (await resp.json()).get("hash")
 
     async def storage_push_file(self, file_content) -> str:
+        """
+        Push a file to the storage service.
+        """
         data = aiohttp.FormData()
         data.add_field("file", file_content)
 

src/aleph/sdk/interface.py

@@ -17,7 +17,12 @@
     Union,
 )
 
-from aleph_message.models import AlephMessage, MessagesResponse, MessageType
+from aleph_message.models import (
+    AlephMessage,
+    MessagesResponse,
+    MessageType,
+    PostMessage,
+)
 from aleph_message.models.program import Encoding
 from aleph_message.status import MessageStatus
 
@@ -89,6 +94,53 @@ async def get_posts(
         """
         pass
 
+    async def get_posts_iterator(
+        self,
+        types: Optional[Iterable[str]] = None,
+        refs: Optional[Iterable[str]] = None,
+        addresses: Optional[Iterable[str]] = None,
+        tags: Optional[Iterable[str]] = None,
+        hashes: Optional[Iterable[str]] = None,
+        channels: Optional[Iterable[str]] = None,
+        chains: Optional[Iterable[str]] = None,
+        start_date: Optional[Union[datetime, float]] = None,
+        end_date: Optional[Union[datetime, float]] = None,
+    ) -> AsyncIterable[PostMessage]:
+        """
+        Fetch all filtered posts, returning an async iterator and fetching them page by page. Might return duplicates
+        but will always return all posts.
+
+        :param types: Types of posts to fetch (Default: all types)
+        :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)
+        :param hashes: Specific item_hashes to fetch
+        :param channels: Channels of the posts to fetch (Default: all channels)
+        :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
+        """
+        total_items = None
+        per_page = self.get_posts.__kwdefaults__["pagination"]
+        page = 1
+        while total_items is None or page * per_page < total_items:
+            resp = await self.get_posts(
+                page=page,
+                types=types,
+                refs=refs,
+                addresses=addresses,
+                tags=tags,
+                hashes=hashes,
+                channels=channels,
+                chains=chains,
+                start_date=start_date,
+                end_date=end_date,
+            )
+            total_items = resp["pagination_total"]
+            page += 1
+            for post in resp["posts"]:
+                yield post
+
     @abstractmethod
     async def download_file(
         self,
@@ -143,6 +195,59 @@ async def get_messages(
         """
         pass
 
+    async def get_messages_iterator(
+        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,
+        hashes: Optional[Iterable[str]] = None,
+        channels: Optional[Iterable[str]] = None,
+        chains: Optional[Iterable[str]] = None,
+        start_date: Optional[Union[datetime, float]] = None,
+        end_date: Optional[Union[datetime, float]] = None,
+    ) -> AsyncIterable[AlephMessage]:
+        """
+        Fetch all filtered messages, returning an async iterator and fetching them page by page. Might return duplicates
+        but will always return all messages.
+
+        :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 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)
+        :param hashes: Specific item_hashes to fetch
+        :param channels: Channels of the posts to fetch (Default: all channels)
+        :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
+        """
+        total_items = None
+        per_page = self.get_messages.__kwdefaults__["pagination"]
+        page = 1
+        while total_items is None or page * per_page < total_items:
+            resp = await self.get_messages(
+                page=page,
+                message_type=message_type,
+                content_types=content_types,
+                content_keys=content_keys,
+                refs=refs,
+                addresses=addresses,
+                tags=tags,
+                hashes=hashes,
+                channels=channels,
+                chains=chains,
+                start_date=start_date,
+                end_date=end_date,
+            )
+            total_items = resp.pagination_total
+            page += 1
+            for message in resp.messages:
+                yield message
+
     @abstractmethod
     async def get_message(
         self,
@@ -191,40 +296,6 @@ def watch_messages(
 
 
 class AuthenticatedAlephClientInterface(AlephClientInterface):
-    @abstractmethod
-    async def ipfs_push(self, content: Mapping) -> str:
-        """
-        Push arbitrary content as JSON to the IPFS service.
-
-        :param content: The dict-like content to upload
-        """
-        pass
-
-    @abstractmethod
-    async def storage_push(self, content: Mapping) -> str:
-        """
-        Push arbitrary content as JSON to the storage service.
-
-        :param content: The dict-like content to upload
-        """
-        pass
-
-    @abstractmethod
-    async def ipfs_push_file(self, file_content: Union[str, bytes]) -> str:
-        """
-        Push a file to the IPFS service.
-
-        :param file_content: The file content to upload
-        """
-        pass
-
-    @abstractmethod
-    async def storage_push_file(self, file_content: Union[str, bytes]) -> str:
-        """
-        Push a file to the storage service.
-        """
-        pass
-
     @abstractmethod
     async def create_post(
         self,