feat: add port for the download_manager package as a replacement to Pooch

Author: ecarrenolozano

ontograph/downloader.py

@@ -213,8 +213,136 @@ def _get_resource_url(
 class DownloadManagerAdapter(DownloaderPort):
     """Alternative downloader implementation.
 
-    Placeholder class for a implement the adapter using the
-    `downloader-manager` by Saezlab.
+    Adapter for the `download_manager` package by Saezlab.
     """
 
-    pass
+    def __init__(
+        self,
+        cache_dir: Path,
+        *,
+        backend: str = 'requests',
+        **kwargs: object,
+    ) -> None:
+        """Initialize the download-manager adapter.
+
+        Args:
+            cache_dir: Directory to store downloaded files.
+            backend: Backend for download_manager ('requests' or 'curl').
+            **kwargs: Extra keyword args forwarded to DownloadManager.
+        """
+        try:
+            import download_manager as dm
+        except ModuleNotFoundError as exc:
+            raise ModuleNotFoundError(
+                'download_manager is not installed. '
+                'Install it to use DownloadManagerAdapter.'
+            ) from exc
+
+        self._cache_dir = cache_dir
+        self._cache_dir.mkdir(parents=True, exist_ok=True)
+        self._manager = dm.DownloadManager(
+            path=str(self._cache_dir),
+            backend=backend,
+            **kwargs,
+        )
+        self._resources_paths: dict[str, Path] = {}
+
+    def get_paths(self) -> dict[str, Path]:
+        """Get paths of all downloaded resources.
+
+        Returns:
+            dict[str, Path]: dictionary mapping resource IDs to file paths
+        """
+        return self._resources_paths
+
+    def fetch_from_url(self, url_ontology: str, filename: str | None) -> Path:
+        """Download an ontology file from a specified URL.
+
+        Args:
+            url_ontology: URL pointing to the ontology file
+            filename: Name to save the file as
+
+        Returns:
+            Path: Path to the downloaded file
+
+        Raises:
+            ValueError: If the URL or filename is empty
+            RequestException: If the download fails
+            IOError: If saving the file fails
+        """
+        self._validate_download_parameters(url_ontology, filename)
+
+        dest = self._cache_dir / filename
+        logging.info(f'Downloading ontology from {url_ontology} as {dest}')
+        result_path = self._manager.download(url_ontology, dest=str(dest))
+        if not result_path:
+            raise OSError('Download manager did not return a file path.')
+
+        result = Path(result_path)
+        self._resources_paths[dest.stem] = result
+        return result
+
+    def fetch_from_catalog(
+        self, resources: list[dict[str, str]], catalog: CatalogOntologies
+    ) -> dict[str, Path]:
+        """Download multiple ontology files defined in a catalog.
+
+        Args:
+            resources: list of dictionaries with resource information
+            catalog: Catalog object containing download URLs
+
+        Returns:
+            dict[str, Path]: dictionary mapping resource IDs to file paths
+
+        Raises:
+            ValueError: If the resources list is empty or URL not found
+            KeyError: If a resource is missing required fields
+        """
+        if not resources:
+            raise ValueError('Resources list for batch download is empty.')
+
+        results = {}
+        for resource in resources:
+            name_id, format_type = self._extract_resource_info(resource)
+            url = self._get_resource_url(name_id, format_type, catalog)
+
+            filename = f'{name_id}.{format_type}'
+            local_path = self.fetch_from_url(
+                url_ontology=url, filename=filename
+            )
+            results[name_id] = local_path
+
+        self._resources_paths.update(results)
+        return results
+
+    def _validate_download_parameters(
+        self, url_ontology: str, filename: str | None
+    ) -> None:
+        if not url_ontology or not url_ontology.strip():
+            raise ValueError('URL cannot be empty')
+
+        if not filename or not filename.strip():
+            raise ValueError('Filename cannot be empty')
+
+    def _extract_resource_info(
+        self, resource: dict[str, str]
+    ) -> tuple[str, str]:
+        name_id = resource.get('name_id')
+        if not name_id:
+            raise KeyError("Resource dictionary must contain 'name_id' key")
+
+        format_type = resource.get(
+            'format', DEFAULT_FORMAT_ONTOLOGY
+        )  # Default to OBO format
+        return name_id, format_type
+
+    def _get_resource_url(
+        self, name_id: str, format_type: str, catalog: CatalogOntologies
+    ) -> str:
+        url = catalog.get_download_url(name_id, format_type)
+        if not url:
+            raise ValueError(
+                f'Cannot find download URL for ontology {name_id} '
+                f'in format {format_type}'
+            )
+        return url