Fixed multiple process running with the api

Author: Lucashsmello

datamint/api/base_api.py

@@ -15,6 +15,7 @@
 import asyncio
 from medimgkit.format_detection import GZIP_MIME_TYPES, DEFAULT_MIME_TYPE, guess_typez, guess_extension
 from datamint.utils.env import ensure_asyncio_loop
+import os
 
 if TYPE_CHECKING:
     from datamint.api.client import Api
@@ -261,6 +262,7 @@ def _stream_request(self, method: str, endpoint: str, **kwargs):
                 for chunk in response.iter_bytes():
                     process_chunk(chunk)
         """
+        self._ensure_client_fresh()
         url = endpoint.lstrip('/')  # Remove leading slash for httpx
 
         try:
@@ -269,6 +271,36 @@ def _stream_request(self, method: str, endpoint: str, **kwargs):
             logger.error(f"Request error for streaming {method} {endpoint}: {e}")
             raise
 
+    def _ensure_client_fresh(self) -> None:
+        """Recreate the HTTP client if the process has been forked.
+
+        PyTorch DataLoader forks worker processes that inherit the parent's
+        httpx.Client with its open connection pool. Those inherited connections
+        are in a corrupt state (the server sees the TCP socket closed/reused
+        and replies with chunked data the worker misinterprets as an HTTP
+        status line). Detecting a PID change and recreating the client
+        avoids this issue.
+        """
+        current_pid = os.getpid()
+        if current_pid != self._pid:
+            logger.debug(
+                f"PID changed from {self._pid} to {current_pid} (DataLoader fork detected). "
+                "Recreating httpx client."
+            )
+            # Close the inherited client without going through our normal
+            # close() path (which would also close the aiohttp session).
+            try:
+                self.client.close()
+            except Exception:
+                pass
+            self.client = BaseApi._create_client(self.config)
+            self._owns_client = True
+            self._pid = current_pid
+            # Invalidate any inherited aiohttp session as well.
+            self._aiohttp_session = None
+            self._aiohttp_connector = None
+
+
     def _make_request(self, method: str, endpoint: str, **kwargs) -> httpx.Response:
         """Make HTTP request with error handling and retries.
 
@@ -283,6 +315,7 @@ def _make_request(self, method: str, endpoint: str, **kwargs) -> httpx.Response:
         Raises:
             httpx.HTTPStatusError: If the request fails
         """
+        self._ensure_client_fresh()
         url = endpoint.lstrip('/')  # Remove leading slash for httpx
 
         if logger.isEnabledFor(logging.DEBUG):

pyproject.toml

@@ -1,7 +1,7 @@
 [project]
 name = "datamint"
 description = "A library for interacting with the Datamint API, designed for efficient data management, processing and Deep Learning workflows."
-version = "2.10.7"
+version = "2.10.8"
 dynamic = ["dependencies"]
 requires-python = ">=3.10"
 readme = "README.md"