feat(client): upload dataset items to data service via http using client

python/lib/client/dmod/client/_version.py

@@ -1 +1 @@
-__version__ = '0.8.2'
+__version__ = '0.8.3'

python/lib/client/dmod/client/dmod_client.py

@@ -9,7 +9,7 @@
 from .request_clients import DataServiceClient, JobClient
 from .client_config import ClientConfig
 from pathlib import Path
-from typing import List, Optional, Type, Union
+from typing import List, Literal, Optional, Type, Union
 
 from functools import reduce
 from .domain_detectors import ClientDataCollectionDomainDetector, ClientUniversalItemDomainDetector
@@ -174,7 +174,11 @@ def _get_transport_client(self, **kwargs) -> TransportLayerClient:
     def client_config(self) -> ClientConfig:
         return self._client_config
 
-    async def data_service_action(self, action: str, **kwargs) -> ResultIndicator:
+    async def data_service_action(
+        self,
+        action: Literal["create", "delete", "upload", "download", "list", "items", "state"],
+        **kwargs,
+    ) -> ResultIndicator:
         """
         Perform a supported data service action.
 
@@ -222,10 +226,17 @@ async def data_service_action(self, action: str, **kwargs) -> ResultIndicator:
     @property
     def data_service_client(self) -> DataServiceClient:
         if self._data_service_client is None:
+            # NOTE: request service is bypassed if data service config has been specified and is active
             if self.client_config.data_service is not None and self.client_config.data_service.active:
                 t_client_type = determine_transport_client_type(self.client_config.data_service.endpoint_protocol)
                 t_client = t_client_type(**self.client_config.data_service.dict())
-                self._data_service_client = DataServiceClient(t_client, self._auth_client)
+                # use http client where applicable when bypassing request service
+                import aiohttp
+                host = self.client_config.data_service.endpoint_host
+                port = self.client_config.data_service.endpoint_port
+                proto = "http" if t_client.ssl_context is None else "https"
+                http_client = aiohttp.ClientSession(f"{proto}://{host}:{port}")
+                self._data_service_client = DataServiceClient(t_client, self._auth_client, http_client=http_client)
             else:
                 self._data_service_client = DataServiceClient(self._get_transport_client(), self._auth_client)
         return self._data_service_client

python/lib/client/dmod/client/request_clients.py

@@ -1,4 +1,7 @@
 from abc import ABC, abstractmethod
+import aiohttp
+import mimetypes
+import ssl
 from dmod.communication import (AuthClient, InvalidMessageResponse, ManagementAction, NGENRequest, NGENRequestResponse,
                                 NgenCalibrationRequest, NgenCalibrationResponse, TransportLayerClient)
 from dmod.communication.client import ConnectionContextClient
@@ -351,6 +354,58 @@ def uses_auth(self) -> bool:
         pass
 
 
+class HttpDataTransferAgent(DataTransferAgent):
+    def __init__(self, http_client: aiohttp.ClientSession, ssl_context: Optional[ssl.SSLContext] = None):
+        self.http_client = http_client
+        self.ssl_context = ssl_context
+
+    async def download_dataset_item(self, dataset_name: str, item_name: str, dest: Path):
+        raise NotImplemented
+
+    async def upload_dataset_item(
+        self, dataset_name: str, item_name: str, source: Path
+    ) -> DatasetManagementResponse:
+        if not source.is_file():
+            return DatasetManagementResponse(
+                success=False,
+                reason="Dataset Upload File Not Found",
+                message=f"File {source!s} does not exist",
+            )
+
+        content_type, _ = mimetypes.guess_type(source)
+        form_data = aiohttp.FormData()
+        with source.open() as fp:
+            form_data.add_field(
+                "obj",
+                fp,
+                filename=item_name,
+                content_type=content_type,
+            )
+            response = await self.http_client.post(
+            "/add_object",
+            data=form_data,
+            params={
+                "dataset_name": dataset_name,
+                "object_name": item_name,
+            },
+            ssl=self.ssl_context,
+        )
+        if not response.ok:
+            return DatasetManagementResponse(
+                success=False,
+                reason="Dataset Item Upload Failed",
+                message=f"Status Code: {response.status}",
+            )
+
+        return DatasetManagementResponse(success=True, message="Dataset Item Upload Successful", reason="Success")
+
+    @property
+    def uses_auth(self) -> bool:
+        # NOTE: At this point in time, the dataset service's rest endpoints do
+        # not support auth
+        return False
+
+
 class SimpleDataTransferAgent(DataTransferAgent):
 
     def __init__(self, transport_client: TransportLayerClient, auth_client: Optional[AuthClient] = None, *args, **kwargs):
@@ -559,10 +614,18 @@ def extract_dataset_names(cls, response: DatasetManagementResponse) -> List[str]
         else:
             return response.data.datasets
 
-    def __init__(self, transport_client: TransportLayerClient, auth_client: Optional[AuthClient] = None, *args, **kwargs):
+    def __init__(
+        self,
+        transport_client: TransportLayerClient,
+        auth_client: Optional[AuthClient] = None,
+        *args,
+        http_client: Optional[aiohttp.ClientSession] = None,
+        **kwargs,
+    ):
         super().__init__(*args, **kwargs)
         self._transport_client: TransportLayerClient = transport_client
         self._auth_client: Optional[AuthClient] = auth_client
+        self._http_client: Optional[aiohttp.ClientSession] = http_client
 
     async def _process_request(self, request: DatasetManagementMessage) -> DatasetManagementResponse:
         """
@@ -918,7 +981,11 @@ async def upload_to_dataset(self, dataset_name: str, paths: Union[Path, List[Pat
             An indicator of whether uploading was successful
         """
         # TODO: see if we can perhaps have multiple agents and thread pool if multiplexing is available
-        tx_agent = SimpleDataTransferAgent(transport_client=self._transport_client, auth_client=self._auth_client)
+        # NOTE: prefer http client if available
+        if self._http_client is not None:
+            tx_agent = HttpDataTransferAgent(http_client=self._http_client, ssl_context=self._transport_client.ssl_context)
+        else:
+            tx_agent = SimpleDataTransferAgent(transport_client=self._transport_client, auth_client=self._auth_client)
         if isinstance(paths, Path):
             paths = [paths]
 

python/lib/client/setup.py

@@ -22,7 +22,7 @@
     license='',
     include_package_data=True,
     #install_requires=['websockets', 'jsonschema'],vi
-    install_requires=['dmod-core>=0.16.0', 'websockets>=8.1', 'pydantic>=1.10.8,~=1.10', 'dmod-communication>=0.19.1',
+    install_requires=['dmod-core>=0.16.0', 'websockets>=8.1', 'pydantic>=1.10.8,~=1.10', 'dmod-communication>=0.19.2',
                       'dmod-externalrequests>=0.6.0', 'dmod-modeldata>=0.12.0'],
     packages=find_namespace_packages(include=['dmod.*'], exclude=['dmod.test'])
 )

python/lib/communication/dmod/communication/_version.py

@@ -1 +1 @@
-__version__ = '0.19.1'
+__version__ = '0.19.2'

python/lib/communication/dmod/communication/client.py

@@ -176,6 +176,18 @@ async def async_recv(self) -> str:
         """
         pass
 
+    @property
+    def ssl_context(self) -> Optional[ssl.SSLContext]:
+        """
+        Client ssl context, if present.
+
+        Returns
+        -------
+        Optional[ssl.SSLContext]
+            Configured ssl context
+        """
+        return self._client_ssl_context
+
 
 class AuthClient:
     """