Add CKAN file source with import and export support #23116
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,284 @@ | ||
| from datetime import datetime | ||
| from typing import ( | ||
| Any, | ||
| ) | ||
| from urllib.parse import urlparse | ||
|
|
||
| import requests | ||
| from fsspec import AbstractFileSystem | ||
|
|
||
| from galaxy.files.models import ( | ||
| AnyRemoteEntry, | ||
| FilesSourceRuntimeContext, | ||
| ) | ||
| from galaxy.files.sources._fsspec import ( | ||
| CacheOptionsDictType, | ||
| FsspecBaseFileSourceConfiguration, | ||
| FsspecBaseFileSourceTemplateConfiguration, | ||
| FsspecFilesSource, | ||
| ) | ||
| from galaxy.util import DEFAULT_SOCKET_TIMEOUT | ||
| from galaxy.util.config_templates import TemplateExpansion | ||
|
|
||
|
|
||
| class CKANFileSystem(AbstractFileSystem): | ||
| def __init__(self, base_url: str, token: str | None = None, **kwargs: Any): | ||
| super().__init__(**kwargs) | ||
| self.base_url = base_url.rstrip("/") # prevents double slashes in URLs | ||
| self.token = token | ||
|
|
||
| def _get_request_headers(self) -> dict[str, str]: | ||
| headers = {} | ||
| # auth header, only if token is provided | ||
| if self.token: | ||
| headers["Authorization"] = self.token # raw api token, CKAN doesnt need bearer prefix | ||
| return headers | ||
|
|
||
| def _raise_for_ckan_error(self, response: requests.Response) -> None: | ||
| if response.status_code in (401, 403): | ||
| raise PermissionError(f"Access denied (HTTP {response.status_code}).") | ||
| if response.status_code == 404: | ||
| raise FileNotFoundError(f"Not found (HTTP {response.status_code}): {response.url}") | ||
| response.raise_for_status() # any other 4xx/5xx still raises a generic HTTPError | ||
|
|
||
| # call ckan action api and return response result | ||
| def _get_response( | ||
| self, | ||
| action: str, | ||
| params: dict[str, Any] | None = None, | ||
| ) -> Any: | ||
| url = f"{self.base_url}/api/3/action/{action}" | ||
| headers = self._get_request_headers() | ||
| response = requests.get(url, headers=headers, timeout=DEFAULT_SOCKET_TIMEOUT, params=params) | ||
| self._raise_for_ckan_error(response) # raise exception for HTTP errors | ||
| payload = response.json() | ||
| # ckan could return HTTP 200 even for errors, so check success field in payload | ||
| if not payload.get("success", False): # success is False or missing | ||
| error = payload.get("error", {}) | ||
| raise Exception(f"CKAN API call failed: {error.get('message') or error}") | ||
| return payload["result"] | ||
|
|
||
| # upload a file to an existing dataset as new resource | ||
| def _post_resource(self, dataset_id: str, resource_name: str, file_path: str) -> None: | ||
| url = f"{self.base_url}/api/3/action/resource_create" | ||
| headers = self._get_request_headers() | ||
| with open(file_path, "rb") as f: | ||
| response = requests.post( | ||
| url, | ||
| headers=headers, | ||
| data={"package_id": dataset_id, "name": resource_name}, # target dataset and display name | ||
| files={"upload": (resource_name, f)}, # filename for download and file content | ||
| timeout=DEFAULT_SOCKET_TIMEOUT, | ||
| ) | ||
| self._raise_for_ckan_error(response) # raise exception for HTTP errors | ||
| payload = response.json() | ||
| # ckan could return HTTP 200 even for errors, so check success field in payload | ||
| if not payload.get("success", False): # success is False or missing | ||
| error = payload.get("error", {}) | ||
| raise Exception(f"CKAN API call failed: {error.get('message') or error}") | ||
|
|
||
| def _list_public_datasets(self) -> list[str]: | ||
| # cheap call that returns just the names of public datasets | ||
| return self._get_response("package_list") | ||
|
|
||
| def _list_private_datasets(self) -> list[str]: | ||
| # for listing private datasets, package_search is needed which returns all metadata, not just names | ||
| # since package_search paginates results, its needed to loop until all datasets are retrieved | ||
| names = [] | ||
| start = 0 | ||
| while True: | ||
| # if no private datasets just returns empty list | ||
| result = self._get_response( | ||
| "package_search", | ||
| params={ | ||
| "fq": "private:true", # filter query to return only private datasets | ||
| "include_private": True, # without this CKAN hides private datasets | ||
| "rows": 1000, # rows is page size, 1000 is max allowed in default CKAN | ||
| "start": start, # pagination offset | ||
| }, | ||
| ) | ||
| names.extend([dataset["name"] for dataset in result["results"]]) | ||
| # result["count"] is total number of private datasets, tells if all are retrieved | ||
| if len(names) >= result["count"]: | ||
| break | ||
| start += 1000 | ||
| return names | ||
|
|
||
| # lists all datasets that are available to the user | ||
| # users private datasets are listed first, then the public ones | ||
| def _list_all_datasets(self) -> list[str]: | ||
| return self._list_private_datasets() + self._list_public_datasets() | ||
|
|
||
| # fetch dataset metadata | ||
| def _get_dataset(self, dataset_id: str) -> dict[str, Any]: | ||
| return self._get_response("package_show", params={"id": dataset_id}) | ||
|
|
||
| # extract resources from dataset payload | ||
| def _list_resources(self, dataset_id: str) -> list[dict[str, Any]]: | ||
| dataset = self._get_dataset(dataset_id) | ||
| return dataset.get("resources", []) | ||
|
|
||
| def _is_root(self, path: str) -> bool: | ||
| return path in ("", "/") | ||
|
|
||
| # in case resource name is missing, use id | ||
| def _resource_name(self, resource: dict[str, Any]) -> str | None: | ||
| return resource.get("name") or resource.get("id") | ||
|
|
||
| # returns last modified as a date object, or None if missing/invalid | ||
| def _parse_modified(self, value: str | None) -> datetime | None: | ||
| if not value: | ||
| return None | ||
| try: | ||
| return datetime.fromisoformat(value.replace("Z", "+00:00")) | ||
| except ValueError: | ||
| return None | ||
|
|
||
| # returns resource entry with optional metadata | ||
| def _resource_entry(self, dataset_id: str, resource: dict[str, Any]) -> dict[str, Any]: | ||
| size = resource.get("size") or resource.get("filesize") | ||
| entry = { | ||
| "name": f"/{dataset_id}/{self._resource_name(resource)}", | ||
| "type": "file", | ||
| "modified": self._parse_modified(resource.get("last_modified") or resource.get("metadata_modified")), | ||
| "id": resource.get("id"), | ||
| } | ||
| # size isnt always known, so this prevents int(None) error | ||
| if size is not None: | ||
| entry["size"] = size | ||
| return entry | ||
|
|
||
| # returns dataset entry with optional metadata | ||
| def _dataset_entry(self, name: str, dataset: dict[str, Any] | None = None) -> dict[str, Any]: | ||
| entry: dict[str, Any] = {"name": f"/{name}", "type": "directory", "size": None} | ||
| if dataset: | ||
| # adds these parameter in case detail=True | ||
| entry["modified"] = self._parse_modified(dataset.get("metadata_modified") or dataset.get("last_modified")) | ||
| entry["id"] = dataset.get("id") | ||
| return entry | ||
|
|
||
| # splits path into dataset_id and optional resource_name | ||
| def _split_path(self, path: str) -> tuple[str, str | None]: | ||
| parts = path.strip("/").split("/", 1) | ||
| dataset_id = parts[0] | ||
| resource_name = None # path is /dataset | ||
| if len(parts) > 1: | ||
| resource_name = parts[1] # path is /dataset/resource | ||
| return dataset_id, resource_name | ||
|
|
||
| # find a resource by name in a dataset | ||
| def _find_resource(self, dataset_id: str, resource_name: str) -> dict[str, Any]: | ||
| resources = self._list_resources(dataset_id) | ||
| for resource in resources: | ||
| if self._resource_name(resource) == resource_name: | ||
| return resource | ||
| raise FileNotFoundError(f"/{dataset_id}/{resource_name}") | ||
|
|
||
| def _is_same_host(self, url: str) -> bool: | ||
| return urlparse(url).hostname == urlparse(self.base_url).hostname | ||
|
|
||
| # list datasets in root or resources in a dataset | ||
| def ls(self, path: str = "", detail: bool = True, **kwargs: Any) -> list[dict[str, Any]] | list[str]: | ||
| if not self._is_root(path): | ||
| # list resources in dataset | ||
| dataset_id, _ = self._split_path(path) # only need dataset_id, resource_name isnt needed | ||
| resources = self._list_resources(dataset_id) | ||
| if detail: | ||
| return [self._resource_entry(dataset_id, r) for r in resources] | ||
| return [self._resource_entry(dataset_id, r)["name"] for r in resources] | ||
|
|
||
| # list datasets in root | ||
| datasets = self._list_all_datasets() | ||
| if detail: | ||
| return [self._dataset_entry(name) for name in datasets] | ||
| return [self._dataset_entry(name)["name"] for name in datasets] | ||
|
|
||
| # get dataset or resource metadata | ||
| def info(self, path: str, **kwargs: Any) -> dict[str, Any]: | ||
| if self._is_root(path): | ||
| return {"name": "/", "type": "directory", "size": None} | ||
| dataset_id, resource_name = self._split_path(path) | ||
| # /dataset | ||
| if not resource_name: | ||
| return self._dataset_entry(dataset_id, self._get_dataset(dataset_id)) | ||
| # /dataset/resource | ||
| resource = self._find_resource(dataset_id, resource_name) | ||
| return self._resource_entry(dataset_id, resource) | ||
|
|
||
| # stream resource content using requests, returns a file-like object | ||
| def _open(self, path: str, mode: str = "rb", **kwargs: Any): | ||
| if mode != "rb": | ||
| raise NotImplementedError("Only read binary mode 'rb' is supported") | ||
| dataset_id, resource_name = self._split_path(path) | ||
| if resource_name is None: | ||
| raise FileNotFoundError(path) | ||
| resource = self._find_resource(dataset_id, resource_name) | ||
| url = resource.get("url") or resource.get("download_url") or None | ||
|
Contributor
There was a problem hiding this comment. Since this |
||
| if not url: | ||
| raise FileNotFoundError(path) | ||
| # only add the auth token if the resource is hosted on the same domain as the CKAN instance | ||
| # to prevent leaking tokens to external URLs | ||
| headers = self._get_request_headers() if self._is_same_host(url) else {} | ||
| response = requests.get(url, stream=True, headers=headers, timeout=DEFAULT_SOCKET_TIMEOUT) | ||
| self._raise_for_ckan_error(response) # raise exception for HTTP errors | ||
| response.raw.decode_content = True | ||
| return response.raw | ||
|
|
||
|
|
||
| class CKANFileSourceTemplateConfiguration(FsspecBaseFileSourceTemplateConfiguration): | ||
| endpoint: str | TemplateExpansion | ||
| token: str | TemplateExpansion | None = None | ||
|
|
||
|
|
||
| class CKANFileSourceConfiguration(FsspecBaseFileSourceConfiguration): | ||
| endpoint: str | ||
| token: str | None = None | ||
|
|
||
|
|
||
| class CKANFilesSource(FsspecFilesSource[CKANFileSourceTemplateConfiguration, CKANFileSourceConfiguration]): | ||
|
Contributor
There was a problem hiding this comment. Really appreciated that you are working on this 🙏 I still think that FsspecFilesSource is not the best abstraction for CKAN. Sorry if I may have suggested that at some point, but I would have expected something else. Since the API is just using requests, I suggest looking at other file source implementations that do exactly that. If trying to fit this into an Thanks for keeping up with the good work! |
||
| plugin_type = "ckan" | ||
| required_module = CKANFileSystem | ||
| required_package = "requests" | ||
|
|
||
| template_config_class = CKANFileSourceTemplateConfiguration | ||
| resolved_config_class = CKANFileSourceConfiguration | ||
|
|
||
| def _open_fs( | ||
| self, context: FilesSourceRuntimeContext[CKANFileSourceConfiguration], cache_options: CacheOptionsDictType | ||
| ) -> CKANFileSystem: | ||
| config = context.config | ||
| return CKANFileSystem(base_url=config.endpoint, token=config.token, **cache_options) | ||
|
|
||
| def _write_from( | ||
| self, target_path: str, native_path: str, context: FilesSourceRuntimeContext[CKANFileSourceConfiguration] | ||
| ) -> None: | ||
| fs = self._open_fs(context, {}) | ||
| dataset_id, resource_name = fs._split_path(target_path) | ||
| if not resource_name: | ||
| raise ValueError("Select a dataset as the upload target, not the root.") | ||
| dataset = fs._get_dataset(dataset_id) | ||
| if not dataset.get("private", False): | ||
| raise ValueError("Cannot export to a public CKAN dataset. Select a private dataset instead.") | ||
| fs._post_resource(dataset_id, resource_name, native_path) | ||
|
|
||
| # for the export destination picker only offer datasets the user can write to | ||
| def _list( | ||
| self, | ||
| context: FilesSourceRuntimeContext[CKANFileSourceConfiguration], | ||
| path="/", | ||
| recursive=False, | ||
| write_intent: bool = False, | ||
| limit: int | None = None, | ||
| offset: int | None = None, | ||
| query: str | None = None, | ||
| sort_by: str | None = None, | ||
| ) -> tuple[list[AnyRemoteEntry], int]: | ||
| if write_intent and path in ("", "/"): | ||
| fs = self._open_fs(context, {}) | ||
| entries = [self._info_to_entry(fs._dataset_entry(name), context.config) for name in fs._list_private_datasets()] | ||
| total_count = len(entries) | ||
| return self._apply_pagination(entries, limit, offset), total_count | ||
| return super()._list(context, path, recursive, write_intent, limit, offset, query, sort_by) | ||
|
|
||
|
|
||
| __all__ = ("CKANFilesSource",) | ||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,35 @@ | ||
| - id: ckan | ||
| version: 0 | ||
| name: CKAN | ||
| description: | | ||
| [CKAN](https://ckan.org) is an open-source data portal platform for managing and sharing datasets. | ||
| This configuration allows you to connect Galaxy to a CKAN instance of your choice. | ||
|
|
||
| **Note:** Your **registered email address** will be used to create datasets in CKAN. | ||
| variables: | ||
| url: | ||
| label: CKAN instance endpoint (e.g. https://demo.ckan.org) | ||
| type: string | ||
| help: | | ||
| The endpoint of the CKAN server you are connecting to. This should be the full URL including the protocol | ||
| (http or https) and the domain name. | ||
| writable: | ||
| label: Allow Galaxy to export data to CKAN? | ||
| type: boolean | ||
| optional: true | ||
| default: true | ||
| help: | | ||
| Allow Galaxy to write data to this CKAN instance. Set it to "Yes" if you want to export data from Galaxy to | ||
| CKAN, set it to "No" if you only need to import data from CKAN to Galaxy. | ||
| secrets: | ||
| token: | ||
| label: CKAN Access Token | ||
| optional: true | ||
| help: | | ||
| The personal access token to use to connect to the CKAN instance. You can generate a new token in your CKAN account settings. | ||
| This will allow Galaxy to access private datasets if you have the necessary permissions. | ||
| configuration: | ||
| type: ckan | ||
| token: "{{ secrets.token }}" | ||
| writable: "{{ variables.writable }}" | ||
| endpoint: "{{ variables.url }}" |
Oops, something went wrong.
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Showing the all public + private datasets for pulling files from a CKAN server is fine but pushing files to
publicdatasets on CKAN should not happen, but only to private / user created datasets.for example, I can pull all these datasets from Harvard's Dataverse repo:
but for pushing, it allows only these datasets that I have access to: