Remove/Refactor: Hacky code for sync clients is hard to maintain; models module is too broad of a name; client module classes had unclear naming

Solution: Restructure and remove the sync wrapper; Move models' functions into new query module and existing files; Rename client classes

Author: hoh

examples/httpgateway.py

@@ -7,7 +7,7 @@
 
 from aleph.sdk.chains.common import get_fallback_private_key
 from aleph.sdk.chains.ethereum import ETHAccount
-from aleph.sdk.client import AuthenticatedAlephClient
+from aleph.sdk.client import AuthenticatedAlephHttpClient
 
 app = web.Application()
 routes = web.RouteTableDef()
@@ -32,7 +32,7 @@ async def source_post(request):
             return web.json_response(
                 {"status": "error", "message": "unauthorized secret"}
             )
-    async with AuthenticatedAlephClient(
+    async with AuthenticatedAlephHttpClient(
         account=app["account"], api_server="https://api2.aleph.im"
     ) as session:
         message, _status = await session.create_post(

examples/metrics.py

@@ -12,7 +12,7 @@
 from aleph_message.status import MessageStatus
 
 from aleph.sdk.chains.ethereum import get_fallback_account
-from aleph.sdk.client import AuthenticatedAlephClient, AuthenticatedUserSessionSync
+from aleph.sdk.client import AuthenticatedAlephClientSync, AuthenticatedAlephHttpClient
 from aleph.sdk.conf import settings
 
 
@@ -54,7 +54,7 @@ def get_cpu_cores():
 
 
 def send_metrics(
-    session: AuthenticatedUserSessionSync, metrics
+    session: AuthenticatedAlephClientSync, metrics
 ) -> Tuple[AlephMessage, MessageStatus]:
     return session.create_aggregate(key="metrics", content=metrics, channel="SYSINFO")
 
@@ -70,7 +70,7 @@ def collect_metrics():
 
 def main():
     account = get_fallback_account()
-    with AuthenticatedAlephClient(
+    with AuthenticatedAlephHttpClient(
         account=account, api_server=settings.API_HOST
     ) as session:
         while True:

examples/mqtt.py

@@ -10,7 +10,7 @@
 
 from aleph.sdk.chains.common import get_fallback_private_key
 from aleph.sdk.chains.ethereum import ETHAccount
-from aleph.sdk.client import AuthenticatedAlephClient
+from aleph.sdk.client import AuthenticatedAlephHttpClient
 from aleph.sdk.conf import settings
 
 
@@ -27,7 +27,7 @@ def get_input_data(value):
 
 
 def send_metrics(account, metrics):
-    with AuthenticatedAlephClient(
+    with AuthenticatedAlephHttpClient(
         account=account, api_server=settings.API_HOST
     ) as session:
         return session.create_aggregate(
@@ -100,7 +100,7 @@ async def gateway(
         if not userdata["received"]:
             await client.reconnect()
 
-        async with AuthenticatedAlephClient(
+        async with AuthenticatedAlephHttpClient(
             account=account, api_server=settings.API_HOST
         ) as session:
             for key, value in state.items():

examples/store.py

@@ -6,7 +6,7 @@
 
 from aleph.sdk.chains.common import get_fallback_private_key
 from aleph.sdk.chains.ethereum import ETHAccount
-from aleph.sdk.client import AuthenticatedAlephClient
+from aleph.sdk.client import AuthenticatedAlephHttpClient
 from aleph.sdk.conf import settings
 
 DEFAULT_SERVER = "https://api2.aleph.im"
@@ -23,7 +23,7 @@ async def print_output_hash(message: StoreMessage, status: MessageStatus):
 
 
 async def do_upload(account, engine, channel, filename=None, file_hash=None):
-    async with AuthenticatedAlephClient(
+    async with AuthenticatedAlephHttpClient(
         account=account, api_server=settings.API_HOST
     ) as session:
         print(filename, account.get_address())

src/aleph/sdk/__init__.py

@@ -1,6 +1,6 @@
 from pkg_resources import DistributionNotFound, get_distribution
 
-from aleph.sdk.client import AlephClient, AuthenticatedAlephClient
+from aleph.sdk.client import AlephHttpClient, AuthenticatedAlephHttpClient
 
 try:
     # Change here if project is renamed and does not equal the package name
@@ -11,4 +11,4 @@
 finally:
     del get_distribution, DistributionNotFound
 
-__all__ = ["AlephClient", "AuthenticatedAlephClient"]
+__all__ = ["AlephHttpClient", "AuthenticatedAlephHttpClient"]

src/aleph/sdk/client/__init__.py

@@ -1,12 +1,10 @@
-from .authenticated import AuthenticatedAlephClient, AuthenticatedUserSessionSync
-from .base import BaseAlephClient, BaseAuthenticatedAlephClient
-from .client import AlephClient, UserSessionSync
+from .abstract import AlephClient, AuthenticatedAlephClient
+from .authenticated_http import AuthenticatedAlephHttpClient
+from .http import AlephHttpClient
 
 __all__ = [
-    "BaseAlephClient",
-    "BaseAuthenticatedAlephClient",
     "AlephClient",
     "AuthenticatedAlephClient",
-    "UserSessionSync",
-    "AuthenticatedUserSessionSync",
+    "AlephHttpClient",
+    "AuthenticatedAlephHttpClient",
 ]

src/aleph/sdk/client/base.py renamed to src/aleph/sdk/client/abstract.py

@@ -25,14 +25,15 @@
 from aleph_message.models.execution.program import Encoding
 from aleph_message.status import MessageStatus
 
-from ..models.message import MessageFilter
-from ..models.post import PostFilter, PostsResponse
+from ..query.filters import MessageFilter, PostFilter
+from ..query.responses import PostsResponse
 from ..types import GenericMessage, StorageEnum
+from ..utils import Writable
 
 DEFAULT_PAGE_SIZE = 200
 
 
-class BaseAlephClient(ABC):
+class AlephClient(ABC):
     @abstractmethod
     async def fetch_aggregate(self, address: str, key: str) -> Dict[str, Dict]:
         """
@@ -110,6 +111,44 @@ async def download_file(
         """
         pass
 
+    async def download_file_ipfs(
+        self,
+        file_hash: str,
+    ) -> bytes:
+        """
+        Get a file from the ipfs storage engine as raw bytes.
+
+        Warning: Downloading large files can be slow.
+
+        :param file_hash: The hash of the file to retrieve.
+        """
+        raise NotImplementedError()
+
+    async def download_file_ipfs_to_buffer(
+        self,
+        file_hash: str,
+        output_buffer: Writable[bytes],
+    ) -> None:
+        """
+        Download a file from the storage engine and write it to the specified output buffer.
+
+        :param file_hash: The hash of the file to retrieve.
+        :param output_buffer: The binary output buffer to write the file data to.
+        """
+        raise NotImplementedError()
+
+    async def download_file_to_buffer(
+        self,
+        file_hash: str,
+        output_buffer: Writable[bytes],
+    ) -> None:
+        """
+        Download a file from the storage engine and write it to the specified output buffer.
+        :param file_hash: The hash of the file to retrieve.
+        :param output_buffer: Writable binary buffer. The file will be written to this buffer.
+        """
+        raise NotImplementedError()
+
     @abstractmethod
     async def get_messages(
         self,
@@ -180,7 +219,7 @@ def watch_messages(
         pass
 
 
-class BaseAuthenticatedAlephClient(BaseAlephClient):
+class AuthenticatedAlephClient(AlephClient):
     @abstractmethod
     async def create_post(
         self,
@@ -350,3 +389,19 @@ async def submit(
         :param sync: If true, waits for the message to be processed by the API server (Default: False)
         """
         pass
+
+    async def ipfs_push(self, content: Mapping) -> str:
+        """
+        Push a file to IPFS.
+
+        :param content: Content of the file to push
+        """
+        raise NotImplementedError()
+
+    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
+        """
+        raise NotImplementedError()