try windows

.github/workflows/lint-unit-tests.yml

@@ -228,48 +228,48 @@ jobs:
         with:
           token: ${{ secrets.CODECOV_TOKEN }}
 
-  #Unit-Test-Windows:
-  #  needs: Lints
-  #  strategy:
-  #    matrix:
-  #      python-version: ["3.11"]
-  #      poetry-version: ["1.7.1"]
-  #      os: [windows]
-  #      node-version: ["22.5.1"]
-  #
-  #  runs-on: ${{ matrix.os }}-2022
-  #  steps:
-  #    - uses: actions/checkout@v4
-  #
-  #    - uses: pnpm/action-setup@v4
-  #      name: Install pnpm
-  #      with:
-  #        version: 9
-  #        run_install: false
-  #
-  #    - name: Use Node.js ${{ matrix.node-version }}
-  #      uses: actions/setup-node@v4
-  #      with:
-  #        node-version: ${{ matrix.node-version }}
-  #        cache: "pnpm"
-  #        cache-dependency-path: "opsml/app/static/pnpm-lock.yaml"
-  #
-  #    - name: Build App for testing
-  #      run: |
-  #        cd opsml/app/static
-  #        pnpm install
-  #        pnpm run build
-  #        pnpm run test
-  #
-  #    - name: Set up uv
-  #      run: irm https://astral.sh/uv/install.ps1 | iex
-  #      shell: powershell
-  #
-  #    - name: Set up Python ${{ matrix.python-version }}
-  #      run: uv python install ${{ matrix.python-version }}
-  #
-  #    - run: make setup.project
-  #    - run: make test.unit
+  Unit-Test-Windows:
+    needs: Lints
+    strategy:
+      matrix:
+        python-version: ["3.11"]
+        poetry-version: ["1.7.1"]
+        os: [windows]
+        node-version: ["22.5.1"]
+
+    runs-on: ${{ matrix.os }}-2019
+    steps:
+      - uses: actions/checkout@v4
+
+      - uses: pnpm/action-setup@v4
+        name: Install pnpm
+        with:
+          version: 9
+          run_install: false
+
+      - name: Use Node.js ${{ matrix.node-version }}
+        uses: actions/setup-node@v4
+        with:
+          node-version: ${{ matrix.node-version }}
+          cache: "pnpm"
+          cache-dependency-path: "opsml/app/static/pnpm-lock.yaml"
+
+      - name: Build App for testing
+        run: |
+          cd opsml/app/static
+          pnpm install
+          pnpm run build
+          pnpm run test
+
+      - name: Set up uv
+        run: irm https://astral.sh/uv/install.ps1 | iex
+        shell: powershell
+
+      - name: Set up Python ${{ matrix.python-version }}
+        run: uv python install ${{ matrix.python-version }}
+
+      - run: make setup.install
+      - run: make test.registry
 
   Postgres-Unit:
     needs: [Unit-Test-Ubuntu, Unit-Test-MacOS, Unit-Test-Coverage]

opsml/app/routes/data.py

@@ -30,17 +30,17 @@
 
 
 @router.get("/data/download", name="download_data")
-def download_data(request: Request, uid: str) -> StreamingResponse:
+async def download_data(request: Request, uid: str) -> StreamingResponse:
     """Downloads data associated with a datacard"""
 
     registry: CardRegistry = request.app.state.registries.data
     datacard = cast(DataCard, registry.load_card(uid=uid))
     load_path = Path(datacard.uri / SaveName.DATA.value).with_suffix(datacard.interface.data_suffix)
-    return download_artifacts_ui(request, str(load_path))
+    return await download_artifacts_ui(request, str(load_path))
 
 
 @router.get("/data/download/profile", name="download_data_profile")
-def download_data_profile(
+async def download_data_profile(
     request: Request,
     uid: str,
 ) -> StreamingResponse:
@@ -49,7 +49,7 @@
     registry: CardRegistry = request.app.state.registries.data
     datacard = cast(DataCard, registry.load_card(uid=uid))
     load_path = Path(datacard.uri / SaveName.DATA_PROFILE.value).with_suffix(Suffix.HTML.value)
-    return download_file(request, str(load_path))
+    return await download_file(request, str(load_path))
 
 
 @router.post("/data/card", name="data_card", response_model=DataCardMetadata)

opsml/app/routes/files.py

@@ -13,6 +13,7 @@
 import streaming_form_data
 from fastapi import APIRouter, Depends, HTTPException, Request, status
 from fastapi.responses import StreamingResponse
+from starlette.concurrency import run_in_threadpool
 from starlette.requests import ClientDisconnect
 from streaming_form_data import StreamingFormDataParser
 from streaming_form_data.validators import MaxSizeValidator
@@ -113,7 +114,7 @@
 
 
 @router.get("/files/download", name="download_file")
-def download_file(request: Request, path: str) -> StreamingResponse:
+async def download_file(request: Request, path: str) -> StreamingResponse:
     """Downloads a file
 
     Args:
@@ -128,13 +129,9 @@
     logger.info("Server: Downloading file {}", path)
     storage_client: StorageClientBase = request.app.state.storage_client
     try:
-        return StreamingResponse(
-            storage_client.iterfile(
-                Path(swap_opsml_root(request, Path(path))),
-                config.download_chunk_size,
-            ),
-            media_type="application/octet-stream",
-        )
+        file_path = Path(swap_opsml_root(request, Path(path)))
+        file_iterator = await run_in_threadpool(storage_client.iterfile, file_path, config.download_chunk_size)
+        return StreamingResponse(file_iterator, media_type="application/octet-stream")
 
     except Exception as error:
         logger.error("Server: Error downloading file {}", path)
@@ -144,7 +141,7 @@
         ) from error
 
 
-def download_dir(request: Request, path: Path) -> StreamingResponse:
+async def download_dir(request: Request, path: Path) -> StreamingResponse:
     """Downloads a file
 
     Args:
@@ -173,15 +170,14 @@
                     curr_rpath = Path(file_)
                     curr_lpath = lpath / curr_rpath.relative_to(rpath)
                     logger.info("Server: Downloading {} to {}", curr_rpath, curr_lpath)
-                    storage_client.get(curr_rpath, curr_lpath)
+                    await run_in_threadpool(storage_client.get, curr_rpath, curr_lpath)
                     zip_filepath = zipfile / curr_rpath.relative_to(rpath)
                     temp_zip.write(curr_lpath, zip_filepath)
 
             logger.info("Server: Sending zip file for {}", path)
-            return StreamingResponse(
-                storage_client.iterbuffer(zip_io, config.download_chunk_size),
-                media_type="application/x-zip-compressed",
-            )
+            iter_buffer = await run_in_threadpool(storage_client.iterbuffer, zip_io, config.download_chunk_size)
+
+            return StreamingResponse(iter_buffer, media_type="application/x-zip-compressed")
 
     except Exception as error:
         raise HTTPException(
@@ -191,7 +187,7 @@
 
 
 @router.get("/files/download/ui", name="download_artifacts")
-def download_artifacts_ui(request: Request, path: str) -> StreamingResponse:
+async def download_artifacts_ui(request: Request, path: str) -> StreamingResponse:
     """Downloads a file
 
     Args:
@@ -204,8 +200,8 @@
         Streaming file response
     """
     if Path(path).suffix == "":
-        return download_dir(request, Path(path))
-    return download_file(request, path)
+        return await download_dir(request, Path(path))
+    return await download_file(request, path)
 
 
 @router.get("/files/list", name="list_files")

opsml/projects/_run_manager.py

@@ -4,11 +4,13 @@
 # This source code is licensed under the MIT license found in the
 # LICENSE file in the root directory of this source tree.
 
+import concurrent
+import concurrent.futures
 import subprocess
 import tempfile
-import threading
 import time
 import uuid
+from concurrent.futures import Future, ThreadPoolExecutor
 from datetime import datetime
 from pathlib import Path
 from queue import Empty, Queue
@@ -108,7 +110,7 @@ def __init__(self, project_info: ProjectInfo, registries: CardRegistries):
         self._project_info = project_info
         self.active_run: Optional[ActiveRun] = None
         self.registries = registries
-        self._hardware_futures: List[threading.Thread] = []
+        self._hardware_futures: List[Future[Any]] = []
 
         run_id = project_info.run_id
         if run_id is not None:
@@ -120,6 +122,16 @@ def __init__(self, project_info: ProjectInfo, registries: CardRegistries):
             self.run_id = None
             self._run_exists = False
 
+        self._thread_executor: Optional[concurrent.futures.ThreadPoolExecutor] = None
+
+    @property
+    def thread_executor(self) -> Optional[concurrent.futures.ThreadPoolExecutor]:
+        return self._thread_executor
+
+    @thread_executor.setter
+    def thread_executor(self, value: concurrent.futures.ThreadPoolExecutor) -> None:
+        self._thread_executor = value
+
     @property
     def project_id(self) -> int:
         assert self._project_info.project_id is not None, "project_id should not be None"
@@ -200,24 +212,17 @@ def _log_hardware_metrics(self, interval: int) -> None:
 
         # run hardware logger in background thread
         queue: Queue[Dict[str, Union[str, datetime, Dict[str, Any]]]] = Queue()
+        self.thread_executor = ThreadPoolExecutor(max_workers=2)
+        assert self.thread_executor is not None, "thread_executor should not be None"
 
+        # submit futures for hardware logging
         self._hardware_futures.append(
-            threading.Thread(
-                target=put_hw_metrics,
-                kwargs={"interval": interval, "run": self.active_run, "queue": queue},
-            )
+            self.thread_executor.submit(put_hw_metrics, interval, self.active_run, queue),
         )
         self._hardware_futures.append(
-            threading.Thread(
-                target=get_hw_metrics,
-                kwargs={"run": self.active_run, "queue": queue},
-            )
+            self.thread_executor.submit(get_hw_metrics, self.active_run, queue),
         )
 
-        for future in self._hardware_futures:
-            future.daemon = True
-            future.start()
-
     def _extract_code(
         self,
         filename: Path,
@@ -350,8 +355,14 @@ def end_run(self) -> None:
         self._run_exists = False
 
         # check if thread executor is still running
-        if self._hardware_futures:
+        if self.thread_executor is not None:
             for future in self._hardware_futures:
-                future.join()
-
+                future.cancel()
+                try:
+                    future.result(timeout=0.1)
+                except Exception:  # pylint: disable=broad-except
+                    pass
+
+            self.thread_executor.shutdown(wait=False)
+            self.thread_executor = None
             self._hardware_futures = []

opsml/storage/api.py

@@ -56,7 +56,7 @@
 
 
 api_routes = ApiRoutes()
-_TIMEOUT_CONFIG = httpx.Timeout(10, read=120, write=120)
+_TIMEOUT_CONFIG = httpx.Timeout(10, read=30, write=30)
 
 
 class ApiClient:
@@ -132,7 +132,11 @@
         """
         try:
             url = f"{self._base_url}/{route}"
-            response = getattr(self.client, request_type.value.lower())(url=url, **kwargs)
+
+            try:
+                response = getattr(self.client, request_type.value.lower())(url=url, **kwargs)
+            except httpx.ReadTimeout as exc:
+                raise ValueError(f"Request timed out for {request_type} request Url: {route}") from exc
 
             if response.status_code == 200:
                 return cast(Dict[str, Any], response.json())
@@ -157,9 +161,13 @@
 
         # self.refresh_token()
         url = f"{self._base_url}/{route}"
-        with self.client.stream(method="POST", url=url, files=files, headers=headers) as response:
-            for data in response.iter_bytes(chunk_size=chunk_size):
-                result += data.decode("utf-8")
+
+        try:
+            with self.client.stream(method="POST", url=url, files=files, headers=headers) as response:
+                for data in response.iter_bytes(chunk_size=chunk_size):
+                    result += data.decode("utf-8")
+        except httpx.ReadTimeout as exc:
+            raise ValueError(f"Request timed out for POST request Url: {route}") from exc
 
         response_result = cast(Dict[str, Any], py_json.loads(result))
 
@@ -184,9 +192,12 @@
         url = f"{self._base_url}/{route}"
 
         with open(local_path.as_posix(), "wb") as local_file:
-            with self.client.stream(method="GET", url=url, params={"path": read_path.as_posix()}) as response:
-                for data in response.iter_bytes(chunk_size=chunk_size):
-                    local_file.write(data)
+            try:
+                with self.client.stream(method="GET", url=url, params={"path": read_path.as_posix()}) as response:
+                    for data in response.iter_bytes(chunk_size=chunk_size):
+                        local_file.write(data)
+            except httpx.ReadTimeout as exc:
+                raise ValueError(f"Request timed out for GET request Url: {route}") from exc
 
         if response.status_code == 200:
             return {"status": 200}  # filler return

pyproject.toml

@@ -8,7 +8,7 @@ dependencies = [
     "fsspec>=2023.1.0, <=2024.2.0",
     "joblib~=1.3",
     "httpx>=0.23.3, <1.0.0",
-    "opsml-cli~=0.5.0",
+    "opsml-cli~=0.6.1",
     "pandas>=1.5.3, <3",
     "polars~=1.0.0",
     "psutil~=5.9.0",

tests/test_interface/test_model_interface_saver_loader_api.py

@@ -29,6 +29,7 @@
 IS_311 = sys.version_info >= (3, 11)
 
 
+@pytest.mark.skipif(WINDOWS_EXCLUDE, reason="skipping")
 def test_save_huggingface_modelcard_api_client(
     huggingface_torch_distilbert: HuggingFaceModel,
     api_storage_client: client.StorageClientBase,

tests/test_projects/test_opsml_project.py

@@ -14,6 +14,7 @@
 from opsml.projects._run_manager import ActiveRunException
 from opsml.projects.active_run import ActiveRun
 from opsml.registry.registry import CardRegistries
+from tests.conftest import WINDOWS_EXCLUDE
 
 
 def test_opsml_artifact_storage(db_registries: CardRegistries) -> None:
@@ -326,6 +327,7 @@ def test_opsml_project_id_creation(db_registries: CardRegistries) -> None:
     assert project.project_id == 1
 
 
+@pytest.mark.skipif(WINDOWS_EXCLUDE, reason="Windows exclude")
 def test_opsml_project_hardware(db_registries: CardRegistries) -> None:
     """verify that we can read artifacts / metrics / cards without making a run
     active."""

tests/test_projects/test_opsml_project_app.py

@@ -8,6 +8,7 @@
 from opsml.projects import OpsmlProject, ProjectInfo
 from opsml.registry.registry import CardRegistries
 from opsml.storage import client
+from tests.conftest import WINDOWS_EXCLUDE
 
 # test_app already performs a few tests with opsml project in client model
 # Adding additional tests here to avoid further cluttering test_app
@@ -105,6 +106,8 @@ def test_opsml_project_log_code_directory(
     )
 
 
+# exclude windows due to file path issues
+@pytest.mark.skipif(WINDOWS_EXCLUDE, reason="Windows exclude")
 def test_opsml_project_hardware_metric(
     test_app: TestClient,
     api_registries: CardRegistries,