✨ Add download-study-logs-fcn (#163)

bisgaard-itis · web-flow · commit 8a210f3cd8e5 · 2024-06-26T11:01:17.000+02:00
* update workflow before publishing python package * fix dependency issue and bump version * point to website in project description * fix broken dependency * improve doc * add github token to download artifacts * ensure only read-access @wvangeit * yet another attempt at downloading artifacts * make sure to use repo that ran the trigger wf * another attempt at fixing * change owner * allow publishing to testpypi also when pr * minor change * revert minor (but breaking) change * minor fix * add debug messages * another debug message * hopefully the final version * final fix * minor fix * move master and tag to individual jobs * add debug messages * dev->post * add python script for determining semantic version * minor changes * minor changes * improve error handling and add version file to artifacts * check if release * minor fix * ensure to enter venv * also when tagging * source venv in publishin workflow * ensure only master * add script for testing 'pure' semver * adapt workflows to new python script * minor change * attempt to evaluate expressions correctly * several fixes to fix tests * ensure repo is checked out in publish workflow * several small fixes * cleanup * debug * minor cleanup * mionr changes * add debug message * minor change * minor change * yet another try * minor change * minor change * minor change * mionr change * minor changes * correct workflow run id * cosmetic change * avoid using gh * change to a single job for publishing * minor cleanup * swap loops in clean up jobs * correction * update server compatibility to new url * minor change to trigger ci * update openapi.json * update openapi.json * add ensure_unique_names fcn * bugfix and improve tsst * minor fixes * minor change * last fixes * add synchronous version of the method * @pcrespov private method and default limits+offsets * add type annotations * @pcrespov add progressbar when downloading logs * add more useful log * minor changes * add pylint exceptions * use pytest.fail @pcrespov * ensure unique file names directly in method * use Optional instead of | * remove superfluous file
diff --git a/api/openapi.json b/api/openapi.json
diff --git a/clients/python/client/osparc/_files_api.py b/clients/python/client/osparc/_files_api.py
@@ -119,7 +119,7 @@ async def upload_file_async(
             configuration=self.api_client.configuration, timeout=timeout_seconds
         ) as session:
             with logging_redirect_tqdm():
-                _logger.info("Uploading %s in %i chunks", file.name, n_urls)
+                _logger.info("Uploading %s in %i chunk(s)", file.name, n_urls)
                 async for chunck, size in tqdm(
                     file_chunk_generator(file, chunk_size),
                     total=n_urls,
@@ -206,15 +206,15 @@ def _search_files(
         sha256_checksum: Optional[str] = None,
         timeout_seconds: int = DEFAULT_TIMEOUT_SECONDS,
     ) -> PaginationGenerator:
-        def pagination_method():
+        def _pagination_method():
             return super(FilesApi, self).search_files_page(
                 file_id=file_id,
                 sha256_checksum=sha256_checksum,
                 _request_timeout=timeout_seconds,
             )
 
         return PaginationGenerator(
-            first_page_callback=pagination_method,
+            first_page_callback=_pagination_method,
             api_client=self.api_client,
             base_url=self.api_client.configuration.host,
             auth=self._auth,
diff --git a/clients/python/client/osparc/_http_client.py b/clients/python/client/osparc/_http_client.py
@@ -88,6 +88,9 @@ async def delete(self, *args, **kwargs) -> httpx.Response:
     async def patch(self, *args, **kwargs) -> httpx.Response:
         return await self._request(self._client.patch, *args, **kwargs)
 
+    async def get(self, *args, **kwargs) -> httpx.Response:
+        return await self._request(self._client.get, *args, **kwargs)
+
     def _wait_callback(self, retry_state: tenacity.RetryCallState) -> int:
         assert retry_state.outcome is not None
         response: httpx.Response = retry_state.outcome.exception().response
diff --git a/clients/python/client/osparc/_solvers_api.py b/clients/python/client/osparc/_solvers_api.py
@@ -1,12 +1,18 @@
 from typing import Any, List, Optional
 
 import httpx
-from osparc_client import OnePageSolverPort, SolverPort
+from osparc_client import JobInputs, OnePageSolverPort, SolverPort
 from osparc_client import SolversApi as _SolversApi
 
 from . import ApiClient
 from ._models import ParentProjectInfo
-from ._utils import PaginationGenerator, dev_feature, dev_features_enabled
+from ._utils import (
+    _DEFAULT_PAGINATION_LIMIT,
+    _DEFAULT_PAGINATION_OFFSET,
+    PaginationGenerator,
+    dev_feature,
+    dev_features_enabled,
+)
 
 
 class SolversApi(_SolversApi):
@@ -60,18 +66,23 @@ def jobs(self, solver_key: str, version: str) -> PaginationGenerator:
             (its "length")
         """
 
-        def pagination_method():
+        def _pagination_method():
             return super(SolversApi, self).get_jobs_page(
-                solver_key=solver_key, version=version, limit=20, offset=0
+                solver_key=solver_key,
+                version=version,
+                limit=_DEFAULT_PAGINATION_LIMIT,
+                offset=_DEFAULT_PAGINATION_OFFSET,
             )
 
         return PaginationGenerator(
-            first_page_callback=pagination_method,
+            first_page_callback=_pagination_method,
             api_client=self.api_client,
             base_url=self.api_client.configuration.host,
             auth=self._auth,
         )
 
-    def create_job(self, solver_key, version, job_inputs, **kwargs):
+    def create_job(
+        self, solver_key: str, version: str, job_inputs: JobInputs, **kwargs
+    ):
         kwargs = {**kwargs, **ParentProjectInfo().model_dump(exclude_none=True)}
         return super().create_job(solver_key, version, job_inputs, **kwargs)
diff --git a/clients/python/client/osparc/_studies_api.py b/clients/python/client/osparc/_studies_api.py
@@ -1,9 +1,24 @@
-from typing import Any
+import asyncio
+import logging
+from pathlib import Path
+from tempfile import mkdtemp
+from typing import Any, Optional
 
+import httpx
+from osparc_client import ApiClient, JobInputs, JobLogsMap, PageStudy
 from osparc_client import StudiesApi as _StudiesApi
+from tqdm.asyncio import tqdm_asyncio
 
+from ._http_client import AsyncHttpClient
 from ._models import ParentProjectInfo
-from ._utils import dev_features_enabled
+from ._utils import (
+    _DEFAULT_PAGINATION_LIMIT,
+    _DEFAULT_PAGINATION_OFFSET,
+    PaginationGenerator,
+    dev_features_enabled,
+)
+
+_logger = logging.getLogger(__name__)
 
 
 class StudiesApi(_StudiesApi):
@@ -24,15 +39,96 @@ class StudiesApi(_StudiesApi):
         "stop_study_job",
     ]
 
+    def __init__(self, api_client: Optional[ApiClient] = None):
+        """Construct object
+
+        Args:
+            api_client (ApiClient, optinal): osparc.ApiClient object
+        """
+        self._super: _StudiesApi = super()
+        self._super.__init__(api_client)
+        user: Optional[str] = self.api_client.configuration.username
+        passwd: Optional[str] = self.api_client.configuration.password
+        self._auth: Optional[httpx.BasicAuth] = (
+            httpx.BasicAuth(username=user, password=passwd)
+            if (user is not None and passwd is not None)
+            else None
+        )
+
     def __getattribute__(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, job_inputs, **kwargs):
+    def create_study_job(self, study_id: str, job_inputs: JobInputs, **kwargs):
         kwargs = {**kwargs, **ParentProjectInfo().model_dump(exclude_none=True)}
         return super().create_study_job(study_id, job_inputs, **kwargs)
 
-    def clone_study(self, study_id, **kwargs):
+    def clone_study(self, study_id: str, **kwargs):
         kwargs = {**kwargs, **ParentProjectInfo().model_dump(exclude_none=True)}
         return super().clone_study(study_id, **kwargs)
+
+    def studies(self) -> PaginationGenerator:
+        def _pagination_method():
+            page_study = super(StudiesApi, self).list_studies(
+                limit=_DEFAULT_PAGINATION_LIMIT, offset=_DEFAULT_PAGINATION_OFFSET
+            )
+            assert isinstance(page_study, PageStudy)  # nosec
+            return page_study
+
+        return PaginationGenerator(
+            first_page_callback=_pagination_method,
+            api_client=self.api_client,
+            base_url=self.api_client.configuration.host,
+            auth=self._auth,
+        )
+
+    def get_study_job_output_logfiles(self, study_id: str, job_id: str) -> Path:
+        return asyncio.run(
+            self.get_study_job_output_logfiles_async(study_id=study_id, job_id=job_id)
+        )
+
+    async def get_study_job_output_logfiles_async(
+        self, study_id: str, job_id: str, download_dir: Optional[Path] = None
+    ) -> Path:
+        """Download study logs. The log from each node will
+        appear as a file with the node's name in the directory"""
+        if download_dir is not None and not download_dir.is_dir():
+            raise RuntimeError(f"{download_dir=} must be a valid directory")
+        logs_map = super().get_study_job_output_logfile(study_id, job_id)
+        assert isinstance(logs_map, JobLogsMap)  # nosec
+        log_links = logs_map.log_links
+        assert log_links  # nosec
+
+        folder = download_dir or Path(mkdtemp()).resolve()
+        assert folder.is_dir()  # nosec
+        async with AsyncHttpClient(
+            configuration=self.api_client.configuration
+        ) as client:
+
+            async def _download(unique_node_name: str, download_link: str) -> None:
+                response = await client.get(download_link)
+                response.raise_for_status()
+                file = folder / unique_node_name
+                ct = 1
+                while file.exists():
+                    file = file.with_stem(f"{file.stem}({ct})")
+                    ct += 1
+                file.touch()
+                for chunk in response.iter_bytes():
+                    file.write_bytes(chunk)
+
+            tasks = [
+                asyncio.create_task(_download(link.node_name, link.download_link))
+                for link in log_links
+            ]
+            _logger.info(
+                "Downloading log files for study_id=%s and job_id=%s...",
+                study_id,
+                job_id,
+            )
+            await tqdm_asyncio.gather(
+                *tasks, disable=(not _logger.isEnabledFor(logging.INFO))
+            )
+
+        return folder
diff --git a/clients/python/client/osparc/_utils.py b/clients/python/client/osparc/_utils.py
@@ -24,6 +24,9 @@
 _MB = _KB * 1024  # in bytes
 _GB = _MB * 1024  # in bytes
 
+_DEFAULT_PAGINATION_LIMIT: int = 20
+_DEFAULT_PAGINATION_OFFSET: int = 0
+
 DEFAULT_TIMEOUT_SECONDS: int = 30 * 60
 
 Page = Union[PageJob, PageFile, PageSolver, PageStudy]
diff --git a/clients/python/test/e2e/conftest.py b/clients/python/test/e2e/conftest.py
@@ -1,12 +1,14 @@
+# pylint: disable=protected-access
 # pylint: disable=redefined-outer-name
+# pylint: disable=too-many-arguments
 # pylint: disable=unused-argument
 # pylint: disable=unused-variable
-# pylint: disable=too-many-arguments
 
 import logging
 import os
 from pathlib import Path
 from typing import Iterable
+from uuid import UUID
 
 import osparc
 import pytest
@@ -71,3 +73,28 @@ def sleeper(api_client: osparc.ApiClient) -> osparc.Solver:
         "simcore/services/comp/itis/sleeper", "2.0.2"
     )  # type: ignore
     return sleeper
+
+
+@pytest.fixture
+def sleeper_study_id(api_client: osparc.ApiClient) -> UUID:
+    """Simple sleeper study template which takes
+    as input a single file containing a single integer"""
+    _test_study_title = "sleeper_test_study"
+    study_api = osparc.StudiesApi(api_client=api_client)
+    for study in study_api.studies():
+        if study.title == _test_study_title:
+            return UUID(study.uid)
+    pytest.fail(f"Could not find {_test_study_title} study")
+
+
+@pytest.fixture
+def file_with_number(
+    tmp_path: Path, api_client: osparc.ApiClient
+) -> Iterable[osparc.File]:
+    files_api = osparc.FilesApi(api_client=api_client)
+    file = tmp_path / "file_with_number.txt"
+    file.write_text("1")
+    server_file = files_api.upload_file(file)
+    yield server_file
+
+    files_api.delete_file(server_file.id)
diff --git a/clients/python/test/e2e/test_files_api.py b/clients/python/test/e2e/test_files_api.py
@@ -1,3 +1,9 @@
+# pylint: disable=protected-access
+# pylint: disable=redefined-outer-name
+# pylint: disable=too-many-arguments
+# pylint: disable=unused-argument
+# pylint: disable=unused-variable
+
 import hashlib
 from pathlib import Path
 
diff --git a/clients/python/test/e2e/test_notebooks.py b/clients/python/test/e2e/test_notebooks.py
@@ -1,3 +1,9 @@
+# pylint: disable=protected-access
+# pylint: disable=redefined-outer-name
+# pylint: disable=too-many-arguments
+# pylint: disable=unused-argument
+# pylint: disable=unused-variable
+
 import shutil
 import sys
 from pathlib import Path
diff --git a/clients/python/test/e2e/test_solvers_api.py b/clients/python/test/e2e/test_solvers_api.py
@@ -1,3 +1,9 @@
+# pylint: disable=protected-access
+# pylint: disable=redefined-outer-name
+# pylint: disable=too-many-arguments
+# pylint: disable=unused-argument
+# pylint: disable=unused-variable
+
 import json
 
 import osparc
@@ -64,7 +70,7 @@ async def test_logstreaming(
     nloglines: int = 0
     url = f"/v0/solvers/{sleeper.id}/releases/{sleeper.version}/jobs/{job.id}/logstream"
     print(f"starting logstreaming from {url}...")
-    
+
     async with async_client.stream(
         "GET",
         url,
diff --git a/clients/python/test/e2e/test_studies.py b/clients/python/test/e2e/test_studies.py
@@ -0,0 +1,63 @@
+# pylint: disable=protected-access
+# pylint: disable=redefined-outer-name
+# pylint: disable=too-many-arguments
+# pylint: disable=unused-argument
+# pylint: disable=unused-variable
+
+import shutil
+from pathlib import Path
+from uuid import UUID
+
+import osparc
+import pytest
+import tenacity
+from _utils import skip_if_osparc_version
+from packaging.version import Version
+
+
+@skip_if_osparc_version(at_least=Version("0.6.6"))
+@pytest.mark.parametrize("download_dir", [True, False])
+async def test_studies_logs(
+    api_client: osparc.ApiClient,
+    file_with_number: osparc.File,
+    sleeper_study_id: UUID,
+    download_dir: bool,
+    tmp_path: Path,
+):
+    studies_api = osparc.StudiesApi(api_client=api_client)
+    job_inputs = osparc.JobInputs(
+        values={
+            "input_file": file_with_number,
+        }
+    )
+    job = studies_api.create_study_job(
+        study_id=f"{sleeper_study_id}", job_inputs=job_inputs
+    )
+    assert isinstance(job, osparc.Job)
+    print(f"Running study job: {job.id}")
+    status = studies_api.start_study_job(study_id=f"{sleeper_study_id}", job_id=job.id)
+    assert isinstance(status, osparc.JobStatus)
+    async for attempt in tenacity.AsyncRetrying(
+        reraise=True,
+        wait=tenacity.wait_fixed(1),
+        stop=tenacity.stop_after_delay(30),
+        retry=tenacity.retry_if_exception_type(AssertionError),
+    ):
+        with attempt:
+            status = studies_api.inspect_study_job(
+                study_id=f"{sleeper_study_id}", job_id=job.id
+            )
+            assert isinstance(status, osparc.JobStatus)
+            assert status.stopped_at is not None
+    assert status.state == "SUCCESS"
+    try:
+        log_dir = await studies_api.get_study_job_output_logfiles_async(
+            study_id=f"{sleeper_study_id}",
+            job_id=job.id,
+            download_dir=tmp_path if download_dir else None,
+        )
+        assert log_dir.is_dir()
+        n_logfiles = sum(1 for _ in log_dir.rglob("*") if _.is_file())
+        assert n_logfiles > 0
+    finally:
+        shutil.rmtree(log_dir)
