Merge pull request #57 from SonanceAI/feat/object-oriented-api

Refactor ResourcesApi and EntityBaseApi for improved structure

Author: Lucashsmello

datamint/api/base_api.py

@@ -1,27 +1,28 @@
 import logging
-from typing import Any, Generator, AsyncGenerator, Sequence
+from typing import Any, Generator, AsyncGenerator, Sequence, TYPE_CHECKING
 import httpx
 from dataclasses import dataclass
 from datamint.exceptions import DatamintException, ResourceNotFoundError
+from datamint.types import ImagingData
 import aiohttp
 import json
-import pydicom.dataset
 from PIL import Image
 import cv2
 import nibabel as nib
-from nibabel.filebasedimages import FileBasedImage as nib_FileBasedImage
 from io import BytesIO
 import gzip
 import contextlib
 import asyncio
-from medimgkit.format_detection import GZIP_MIME_TYPES
+from medimgkit.format_detection import GZIP_MIME_TYPES, DEFAULT_MIME_TYPE, guess_typez, guess_extension
+
+if TYPE_CHECKING:
+    from datamint.api.client import Api
 
 logger = logging.getLogger(__name__)
 
 # Generic type for entities
 _PAGE_LIMIT = 5000
 
-
 @dataclass
 class ApiConfig:
     """Configuration for API client.
@@ -37,6 +38,15 @@ class ApiConfig:
     timeout: float = 30.0
     max_retries: int = 3
 
+    @property
+    def web_app_url(self) -> str:
+        """Get the base URL for the web application."""
+        if self.server_url.startswith('http://localhost:3001'):
+            return 'http://localhost:3000'
+        if self.server_url.startswith('https://stagingapi.datamint.io'):
+            return 'https://staging.datamint.io'
+        return 'https://app.datamint.io'
+
 
 class BaseApi:
     """Base class for all API endpoint handlers."""
@@ -53,6 +63,7 @@ def __init__(self,
         self.config = config
         self.client = client or self._create_client()
         self.semaphore = asyncio.Semaphore(20)
+        self._api_instance: 'Api | None' = None  # Injected by Api class
 
     def _create_client(self) -> httpx.Client:
         """Create and configure HTTP client with authentication and timeouts."""
@@ -399,10 +410,30 @@ def _convert_array_response(self,
 
     @staticmethod
     def convert_format(bytes_array: bytes,
-                       mimetype: str,
+                       mimetype: str | None = None,
                        file_path: str | None = None
-                       ) -> pydicom.dataset.Dataset | Image.Image | cv2.VideoCapture | bytes | nib_FileBasedImage:
-        """ Convert the bytes array to the appropriate format based on the mimetype."""
+                       ) -> ImagingData | bytes:
+        """ Convert the bytes array to the appropriate format based on the mimetype.
+
+        Args:
+            bytes_array: Raw file content bytes
+            mimetype: Optional MIME type of the content
+            file_path: deprecated
+
+        Returns:
+            Converted content in appropriate format (pydicom.Dataset, PIL Image, cv2.VideoCapture, ...)
+
+        Example:
+            >>> fpath = 'path/to/file.dcm'
+            >>> with open(fpath, 'rb') as f:
+            ...     dicom_bytes = f.read()
+            >>> dicom = BaseApi.convert_format(dicom_bytes)
+
+        """
+        if mimetype is None:
+            mimetype, ext = BaseApi._determine_mimetype(bytes_array)
+            if mimetype is None:
+                raise ValueError("Could not determine mimetype from content.")
         content_io = BytesIO(bytes_array)
         if mimetype.endswith('/dicom'):
             return pydicom.dcmread(content_io)
@@ -429,3 +460,30 @@ def convert_format(bytes_array: bytes,
                 return nib.Nifti1Image.from_stream(f)
 
         raise ValueError(f"Unsupported mimetype: {mimetype}")
+
+    @staticmethod
+    def _determine_mimetype(content: bytes,
+                            declared_mimetype: str | None = None) -> tuple[str | None, str | None]:
+        """Infer MIME type and file extension from content and optional declared type.
+
+        Args:
+            content: Raw file content bytes
+            declared_mimetype: Optional MIME type declared by the source
+
+        Returns:
+            Tuple of (inferred_mimetype, file_extension)
+        """
+        # Determine mimetype from file content
+        mimetype_list, ext = guess_typez(content, use_magic=True)
+        mimetype = mimetype_list[-1]
+
+        # get mimetype from resource info if not detected
+        if declared_mimetype is not None:
+            if mimetype is None:
+                mimetype = declared_mimetype
+                ext = guess_extension(mimetype)
+            elif mimetype == DEFAULT_MIME_TYPE:
+                mimetype = declared_mimetype
+                ext = guess_extension(mimetype)
+
+        return mimetype, ext

datamint/api/client.py

@@ -70,7 +70,10 @@ def check_connection(self):
     def _get_endpoint(self, name: str):
         if name not in self._endpoints:
             api_class = self._API_MAP[name]
-            self._endpoints[name] = api_class(self.config, self._client)
+            endpoint = api_class(self.config, self._client)
+            # Inject this API instance into the endpoint so it can inject into entities
+            endpoint._api_instance = self
+            self._endpoints[name] = endpoint
         return self._endpoints[name]
 
     @property

datamint/api/dto/__init__.py

@@ -1,6 +1,9 @@
 from datamint.apihandler.dto import annotation_dto
-from datamint.apihandler.dto.annotation_dto import AnnotationType, CreateAnnotationDto, Geometry, BoxGeometry, LineGeometry, CoordinateSystem
-
+from datamint.apihandler.dto.annotation_dto import (
+    AnnotationType, CreateAnnotationDto,
+    Geometry, BoxGeometry, LineGeometry,
+    CoordinateSystem
+)
 
 __all__ = [
     "annotation_dto",
@@ -10,4 +13,6 @@
     "BoxGeometry",
     "LineGeometry",
     "CoordinateSystem"
-]
+    "LineGeometry",
+    "CoordinateSystem"
+]

datamint/api/endpoints/annotations_api.py

@@ -31,15 +31,21 @@
 class AnnotationsApi(CreatableEntityApi[Annotation], DeletableEntityApi[Annotation]):
     """API handler for annotation-related endpoints."""
 
-    def __init__(self, config: ApiConfig, client: httpx.Client | None = None) -> None:
+    def __init__(self,
+                 config: ApiConfig,
+                 client: httpx.Client | None = None,
+                 models_api=None,
+                 resources_api=None) -> None:
         """Initialize the annotations API handler.
 
         Args:
             config: API configuration containing base URL, API key, etc.
             client: Optional HTTP client instance. If None, a new one will be created.
         """
+        from .resources_api import ResourcesApi
         super().__init__(config, Annotation, 'annotations', client)
-        self._models_api = ModelsApi(config, client=client)
+        self._models_api = ModelsApi(config, client=client) if models_api is None else models_api
+        self._resources_api = ResourcesApi(config, client=client, annotations_api=self) if resources_api is None else resources_api
 
     def get_list(self,
                  resource: str | Resource | None = None,
@@ -934,7 +940,7 @@ def _create_geometry_annotation(self,
 
     def download_file(self,
                       annotation: str | Annotation,
-                      fpath_out: str | Path | None = None) -> bytes:
+                      fpath_out: str | os.PathLike | None = None) -> bytes:
         """
         Download the segmentation file for a given resource and annotation.
 
@@ -954,7 +960,7 @@ def download_file(self,
 
         resp = self._make_request('GET', f'/annotations/{resource_id}/annotations/{annotation_id}/file')
         if fpath_out:
-            with open(str(fpath_out), 'wb') as f:
+            with open(fpath_out, 'wb') as f:
                 f.write(resp.content)
         return resp.content
 
@@ -1059,3 +1065,6 @@ def patch(self,
         respdata = resp.json()
         if isinstance(respdata, dict) and 'error' in respdata:
             raise DatamintException(respdata['error'])
+
+    def _get_resource(self, ann: Annotation) -> Resource:
+        return self._resources_api.get_by_id(ann.resource_id)

datamint/api/endpoints/projects_api.py

@@ -30,7 +30,8 @@ def get_project_resources(self, project: Project | str) -> list[Resource]:
         """
         response = self._get_child_entities(project, 'resources')
         resources_data = response.json()
-        return [Resource(**item) for item in resources_data]
+        resources = [Resource(**item) for item in resources_data]
+        return resources
 
     def create(self,
                name: str,
@@ -148,47 +149,48 @@ def add_resources(self,
         self._make_entity_request('POST', project_id, add_path='resources',
                                   json={'resource_ids_to_add': resources_ids, 'all_files_selected': False})
 
-    def download(self, project: str | Project,
-                 outpath: str,
-                 all_annotations: bool = False,
-                 include_unannotated: bool = False,
-                 ) -> None:
-        """Download a project by its id.
-
-        Args:
-            project: The project id or Project instance.
-            outpath: The path to save the project zip file.
-            all_annotations: Whether to include all annotations in the downloaded dataset,
-                even those not made by the provided project.
-            include_unannotated: Whether to include unannotated resources in the downloaded dataset.
-        """
-        from tqdm.auto import tqdm
-        params = {'all_annotations': all_annotations}
-        if include_unannotated:
-            params['include_unannotated'] = include_unannotated
-
-        project_id = self._entid(project)
-        with self._stream_entity_request('GET', project_id,
-                                         add_path='annotated_dataset',
-                                         params=params) as response:
-            total_size = int(response.headers.get('content-length', 0))
-            if total_size == 0:
-                total_size = None
-            with tqdm(total=total_size, unit='B', unit_scale=True) as progress_bar:
-                with open(outpath, 'wb') as file:
-                    for data in response.iter_bytes(1024):
-                        progress_bar.update(len(data))
-                        file.write(data)
+    # def download(self, project: str | Project,
+    #              outpath: str,
+    #              all_annotations: bool = False,
+    #              include_unannotated: bool = False,
+    #              ) -> None:
+    #     """Download a project by its id.
+
+    #     Args:
+    #         project: The project id or Project instance.
+    #         outpath: The path to save the project zip file.
+    #         all_annotations: Whether to include all annotations in the downloaded dataset,
+    #             even those not made by the provided project.
+    #         include_unannotated: Whether to include unannotated resources in the downloaded dataset.
+    #     """
+    #     from tqdm.auto import tqdm
+    #     params = {'all_annotations': all_annotations}
+    #     if include_unannotated:
+    #         params['include_unannotated'] = include_unannotated
+
+    #     project_id = self._entid(project)
+    #     with self._stream_entity_request('GET', project_id,
+    #                                      add_path='annotated_dataset',
+    #                                      params=params) as response:
+    #         total_size = int(response.headers.get('content-length', 0))
+    #         if total_size == 0:
+    #             total_size = None
+    #         with tqdm(total=total_size, unit='B', unit_scale=True) as progress_bar:
+    #             with open(outpath, 'wb') as file:
+    #                 for data in response.iter_bytes(1024):
+    #                     progress_bar.update(len(data))
+    #                     file.write(data)
 
     def set_work_status(self,
-                        resource: str | Resource,
                         project: str | Project,
+                        resource: str | Resource,
                         status: Literal['opened', 'annotated', 'closed']) -> None:
         """
         Set the status of a resource.
 
         Args:
-            annotation: The annotation unique id or an annotation object.
+            project: The project unique id or a project object.
+            resource: The resource unique id or a resource object.
             status: The new status to set.
         """
         resource_id = self._entid(resource)