merge master into 201-experiment-with-upgrading-openapi-generator

Author: bisgaard-itis

.github/workflows/publish-python-client.yml

@@ -32,5 +32,6 @@ jobs:
       - name: Publish to PyPI
         uses: pypa/gh-action-pypi-publish@release/v1
         with:
+          attestations: false
           verbose: true
           packages-dir: osparc_python_wheels/

clients/python/requirements/e2e-test.txt

@@ -1,10 +1,12 @@
 -r ../../../requirements.txt
 
 black
+faker
 ipykernel
 ipython
 jinja2
 matplotlib
+memory_profiler
 packaging
 pandas
 papermill

clients/python/setup.py

@@ -31,6 +31,7 @@
     "tqdm>=4.48.0",
     f"osparc_client=={VERSION}",
     "urllib3",
+    "aiofiles",
 ]
 
 SETUP = dict(

clients/python/src/osparc/_api_files_api.py

@@ -4,9 +4,6 @@
 import json
 import logging
 import math
-import random
-import shutil
-import string
 from pathlib import Path
 from typing import Any, Iterator, List, Optional, Tuple, Union
 from tempfile import NamedTemporaryFile
@@ -29,11 +26,13 @@
     FileUploadData,
     UploadedPart,
 )
+from urllib.parse import urljoin
+import aiofiles
+import shutil
 from ._utils import (
     DEFAULT_TIMEOUT_SECONDS,
     PaginationGenerator,
     compute_sha256,
-    dev_features_enabled,
     file_chunk_generator,
 )
 
@@ -60,34 +59,58 @@ def __init__(self, api_client: ApiClient):
             else None
         )
 
-    def __getattr__(self, name: str) -> Any:
-        if (name in FilesApi._dev_features) and (not dev_features_enabled()):
-            raise NotImplementedError(f"FilesApi.{name} is still under development")
-        return super().__getattribute__(name)
-
     def download_file(
-        self, file_id: str, *, destination_folder: Optional[Path] = None, **kwargs
+        self,
+        file_id: str,
+        *,
+        destination_folder: Optional[Path] = None,
+        timeout_seconds: int = DEFAULT_TIMEOUT_SECONDS,
+        **kwargs,
+    ) -> str:
+        return asyncio.run(
+            self.download_file_async(
+                file_id=file_id,
+                destination_folder=destination_folder,
+                timeout_seconds=timeout_seconds,
+                **kwargs,
+            )
+        )
+
+    async def download_file_async(
+        self,
+        file_id: str,
+        *,
+        destination_folder: Optional[Path] = None,
+        timeout_seconds: int = DEFAULT_TIMEOUT_SECONDS,
+        **kwargs,
     ) -> str:
         if destination_folder is not None and not destination_folder.is_dir():
             raise RuntimeError(
                 f"destination_folder: {destination_folder} must be a directory"
             )
-        with NamedTemporaryFile(delete=False) as tmp_file:
-            downloaded_file = Path(tmp_file.name)
-        data = super().download_file(file_id, **kwargs)
-        downloaded_file.write_bytes(data)
-        if destination_folder is not None:
-            dest_file: Path = destination_folder / downloaded_file.name
-            while dest_file.is_file():
-                new_name = (
-                    downloaded_file.stem
-                    + "".join(random.choices(string.ascii_letters, k=8))
-                    + downloaded_file.suffix
+        async with aiofiles.tempfile.NamedTemporaryFile(
+            mode="wb",
+            delete=False,
+        ) as downloaded_file:
+            async with AsyncHttpClient(
+                configuration=self.api_client.configuration, timeout=timeout_seconds
+            ) as session:
+                url = urljoin(
+                    self.api_client.configuration.host, f"/v0/files/{file_id}/content"
                 )
-                dest_file = destination_folder / new_name
-            shutil.move(downloaded_file, dest_file)
-            downloaded_file = dest_file
-        return str(downloaded_file.resolve())
+                async for response in await session.stream(
+                    "GET", url=url, auth=self._auth, follow_redirects=True
+                ):
+                    response.raise_for_status()
+                    async for chunk in response.aiter_bytes():
+                        await downloaded_file.write(chunk)
+        dest_file = f"{downloaded_file.name}"
+        if destination_folder is not None:
+            dest_file = NamedTemporaryFile(dir=destination_folder, delete=False).name
+            shutil.move(
+                f"{downloaded_file.name}", dest_file
+            )  # aiofiles doesnt seem to have an async variant of this
+        return dest_file
 
     def upload_file(
         self,
@@ -109,7 +132,7 @@ async def upload_file_async(
             file = Path(file)
         if not file.is_file():
             raise RuntimeError(f"{file} is not a file")
-        checksum: str = compute_sha256(file)
+        checksum: str = await compute_sha256(file)
         for file_result in self._search_files(
             sha256_checksum=checksum, timeout_seconds=timeout_seconds
         ):
@@ -163,7 +186,7 @@ async def upload_file_async(
             )
             async with AsyncHttpClient(
                 configuration=self.api_client.configuration,
-                request_type="post",
+                method="post",
                 url=links.abort_upload,
                 body=abort_body.to_dict(),
                 base_url=self.api_client.configuration.host,

clients/python/src/osparc/_api_solvers_api.py

@@ -1,6 +1,6 @@
 # Wraps osparc_client.api.solvers_api
 
-from typing import Any, List, Optional
+from typing import List, Optional
 
 import httpx
 from osparc_client.api.solvers_api import SolversApi as _SolversApi
@@ -22,8 +22,6 @@
     _DEFAULT_PAGINATION_LIMIT,
     _DEFAULT_PAGINATION_OFFSET,
     PaginationGenerator,
-    dev_feature,
-    dev_features_enabled,
 )
 import warnings
 from tempfile import NamedTemporaryFile
@@ -54,11 +52,6 @@ def __init__(self, api_client: ApiClient):
             else None
         )
 
-    def __getattr__(self, name: str) -> Any:
-        if (name in SolversApi._dev_features) and (not dev_features_enabled()):
-            raise NotImplementedError(f"SolversApi.{name} is still under development")
-        return super().__getattribute__(name)
-
     def list_solver_ports(
         self, solver_key: str, version: str, **kwargs
     ) -> List[SolverPort]:
@@ -67,7 +60,6 @@ def list_solver_ports(
         )
         return page.items if page.items else []
 
-    @dev_feature
     def iter_jobs(self, solver_key: str, version: str, **kwargs) -> PaginationGenerator:
         """Returns an iterator through which one can iterate over
         all Jobs submitted to the solver
@@ -100,7 +92,6 @@ def _pagination_method():
             auth=self._auth,
         )
 
-    @dev_feature
     def jobs(self, solver_key: str, version: str, **kwargs) -> PaginationGenerator:
         warnings.warn(
             "The 'jobs' method is deprecated and will be removed in a future version. "

clients/python/src/osparc/_api_studies_api.py

@@ -4,7 +4,7 @@
 import logging
 from pathlib import Path
 from tempfile import mkdtemp
-from typing import Any, Optional
+from typing import Optional
 
 import httpx
 from pydantic import StrictStr
@@ -29,7 +29,6 @@
     _DEFAULT_PAGINATION_LIMIT,
     _DEFAULT_PAGINATION_OFFSET,
     PaginationGenerator,
-    dev_features_enabled,
 )
 import warnings
 
@@ -69,11 +68,6 @@ def __init__(self, api_client: ApiClient):
             else None
         )
 
-    def __getattr__(self, name: str) -> Any:
-        if (name in StudiesApi._dev_features) and (not dev_features_enabled()):
-            raise NotImplementedError(f"StudiesApi.{name} is still under development")
-        return super().__getattribute__(name)
-
     def create_study_job(self, study_id: str, job_inputs: JobInputs, **kwargs):
         _job_inputs = _JobInputs.from_json(job_inputs.model_dump_json())
         assert _job_inputs is not None

clients/python/src/osparc/_http_client.py

@@ -1,7 +1,16 @@
 from contextlib import suppress
 from datetime import datetime
 from email.utils import parsedate_to_datetime
-from typing import Any, Awaitable, Callable, Dict, Optional, Set
+from typing import (
+    Any,
+    Awaitable,
+    Callable,
+    Dict,
+    Optional,
+    Set,
+    Literal,
+    AsyncGenerator,
+)
 
 import httpx
 import tenacity
@@ -17,14 +26,14 @@ def __init__(
         self,
         *,
         configuration: Configuration,
-        request_type: Optional[str] = None,
+        method: Optional[str] = None,
         url: Optional[str] = None,
         body: Optional[Dict] = None,
         **httpx_async_client_kwargs,
     ):
         self.configuration = configuration
         self._client = httpx.AsyncClient(**httpx_async_client_kwargs)
-        self._callback = getattr(self._client, request_type) if request_type else None
+        self._callback = getattr(self._client, method) if method else None
         self._url = url
         self._body = body
         if self._callback is not None:
@@ -77,6 +86,28 @@ async def _():
 
         return await _()
 
+    async def _stream(
+        self, method: Literal["GET"], url: str, *args, **kwargs
+    ) -> AsyncGenerator[httpx.Response, None]:
+        n_attempts = self.configuration.retries.total
+        assert isinstance(n_attempts, int)
+
+        @tenacity.retry(
+            reraise=True,
+            wait=self._wait_callback,
+            stop=tenacity.stop_after_attempt(n_attempts),
+            retry=tenacity.retry_if_exception_type(httpx.HTTPStatusError),
+        )
+        async def _() -> AsyncGenerator[httpx.Response, None]:
+            async with self._client.stream(
+                method=method, url=url, *args, **kwargs
+            ) as response:
+                if response.status_code in self.configuration.retries.status_forcelist:
+                    response.raise_for_status()
+                yield response
+
+        return _()
+
     async def put(self, *args, **kwargs) -> httpx.Response:
         return await self._request(self._client.put, *args, **kwargs)
 
@@ -92,6 +123,11 @@ async def patch(self, *args, **kwargs) -> httpx.Response:
     async def get(self, *args, **kwargs) -> httpx.Response:
         return await self._request(self._client.get, *args, **kwargs)
 
+    async def stream(
+        self, method: Literal["GET"], url: str, *args, **kwargs
+    ) -> AsyncGenerator[httpx.Response, None]:
+        return await self._stream(method=method, url=url, *args, **kwargs)
+
     def _wait_callback(self, retry_state: tenacity.RetryCallState) -> int:
         assert retry_state.outcome is not None
         if retry_state.outcome and retry_state.outcome.exception():

clients/python/src/osparc/_utils.py

@@ -1,7 +1,5 @@
 import asyncio
 import hashlib
-import os
-from functools import wraps
 from pathlib import Path
 from typing import AsyncGenerator, Callable, Generator, Optional, Tuple, TypeVar, Union
 
@@ -16,7 +14,7 @@
     Solver,
     Study,
 )
-
+import aiofiles
 from ._exceptions import RequestError
 
 _KB = 1024  # in bytes
@@ -87,15 +85,15 @@ async def file_chunk_generator(
     bytes_read: int = 0
     file_size: int = file.stat().st_size
     while bytes_read < file_size:
-        with open(file, "rb") as f:
-            f.seek(bytes_read)
+        async with aiofiles.open(file, "rb") as f:
+            await f.seek(bytes_read)
             nbytes = (
                 chunk_size
                 if (bytes_read + chunk_size <= file_size)
                 else (file_size - bytes_read)
             )
             assert nbytes > 0
-            chunk = await asyncio.get_event_loop().run_in_executor(None, f.read, nbytes)
+            chunk = await f.read(nbytes)
             yield chunk, nbytes
             bytes_read += nbytes
 
@@ -109,27 +107,13 @@ async def _fcn_to_coro(callback: Callable[..., S], *args) -> S:
     return result
 
 
-def compute_sha256(file: Path) -> str:
+async def compute_sha256(file: Path) -> str:
     assert file.is_file()
     sha256 = hashlib.sha256()
-    with open(file, "rb") as f:
+    async with aiofiles.open(file, "rb") as f:
         while True:
-            data = f.read(100 * _KB)
+            data = await f.read(100 * _KB)
             if not data:
                 break
             sha256.update(data)
-        return sha256.hexdigest()
-
-
-def dev_features_enabled() -> bool:
-    return os.environ.get("OSPARC_DEV_FEATURES_ENABLED") == "1"
-
-
-def dev_feature(func: Callable):
-    @wraps(func)
-    def _wrapper(*args, **kwargs):
-        if not dev_features_enabled():
-            raise NotImplementedError(f"{func.__name__} is still under development")
-        return func(*args, **kwargs)
-
-    return _wrapper
+    return sha256.hexdigest()

clients/python/test/e2e/_utils.py

@@ -1,4 +1,3 @@
-import os
 import subprocess
 from pathlib import Path
 from typing import Optional
@@ -11,10 +10,6 @@
 assert _clients_python_dir.is_dir()
 
 
-def osparc_dev_features_enabled() -> bool:
-    return os.environ.get("OSPARC_DEV_FEATURES_ENABLED") == "1"
-
-
 def repo_version() -> Version:
     subprocess.run(
         "make VERSION", cwd=_clients_python_dir.resolve(), shell=True