Source code for galaxy.files.sources.googlecloudstorage

import logging

from galaxy.exceptions import ConfigurationError
from galaxy.files.models import FilesSourceRuntimeContext
from galaxy.files.sources._fsspec import (
    CacheOptionsDictType,
    FsspecBaseFileSourceConfiguration,
    FsspecBaseFileSourceTemplateConfiguration,
    FsspecFilesSource,
)
from galaxy.util.config_templates import TemplateExpansion

try:
    from gcsfs import GCSFileSystem
except ImportError:
    GCSFileSystem = None


REQUIRED_PACKAGE = "gcsfs"

log = logging.getLogger(__name__)


class GoogleCloudStorageFileSourceTemplateConfiguration(FsspecBaseFileSourceTemplateConfiguration):
    bucket_name: str | TemplateExpansion
    root_path: str | TemplateExpansion | None = None
    project: str | TemplateExpansion | None = None
    # True bills `project`; a project id string bills that project instead (gcsfs semantics).
    requester_pays: bool | str | TemplateExpansion | None = False
    anonymous: bool | TemplateExpansion | None = True
    service_account_json: str | TemplateExpansion | None = None
    # OAuth credentials
    client_id: str | TemplateExpansion | None = None
    client_secret: str | TemplateExpansion | None = None
    token: str | TemplateExpansion | None = None
    refresh_token: str | TemplateExpansion | None = None
    token_uri: str | TemplateExpansion | None = "https://oauth2.googleapis.com/token"


class GoogleCloudStorageFileSourceConfiguration(FsspecBaseFileSourceConfiguration):
    bucket_name: str
    root_path: str | None = None
    project: str | None = None
    requester_pays: bool | str | None = False
    anonymous: bool | None = True
    service_account_json: str | None = None
    # OAuth credentials
    client_id: str | None = None
    client_secret: str | None = None
    token: str | None = None
    refresh_token: str | None = None
    token_uri: str | None = "https://oauth2.googleapis.com/token"



[docs]
class GoogleCloudStorageFilesSource(
    FsspecFilesSource[GoogleCloudStorageFileSourceTemplateConfiguration, GoogleCloudStorageFileSourceConfiguration]
):
    plugin_type = "googlecloudstorage"
    required_module = GCSFileSystem
    required_package = REQUIRED_PACKAGE

    template_config_class = GoogleCloudStorageFileSourceTemplateConfiguration
    resolved_config_class = GoogleCloudStorageFileSourceConfiguration

    def _open_fs(
        self,
        context: FilesSourceRuntimeContext[GoogleCloudStorageFileSourceConfiguration],
        cache_options: CacheOptionsDictType,
    ):
        if GCSFileSystem is None:
            raise self.required_package_exception

        config = context.config
        if config.requester_pays:
            # GCS only bills authenticated callers, and gcsfs silently drops a missing userProject.
            if config.anonymous:
                raise ConfigurationError(
                    f"Requester-pays access to {self.label} needs credentials to bill; set anonymous to false."
                )
            if config.requester_pays is True and not config.project:
                raise ConfigurationError(
                    f"Requester-pays access to {self.label} needs a project to bill; set project or pass a "
                    "project id as requester_pays."
                )
        token: str | dict[str, str | None] | None

        if config.anonymous:
            # Use token='anon' for anonymous access to public buckets
            token = "anon"
        elif config.service_account_json:
            # Path to service account JSON file
            token = config.service_account_json
        elif config.token:
            # OAuth credentials passed as a dictionary
            token = {
                "access_token": config.token,
                "refresh_token": config.refresh_token,
                "client_id": config.client_id,
                "client_secret": config.client_secret,
                "token_uri": config.token_uri,
            }
        else:
            # Default: use application default credentials
            token = None

        fs = GCSFileSystem(
            project=config.project,
            token=token,
            requester_pays=config.requester_pays or False,
            **cache_options,
        )
        return fs

    def _to_filesystem_path(self, path: str, config: GoogleCloudStorageFileSourceConfiguration) -> str:
        """Convert an entry path to the GCS filesystem path format."""
        bucket = config.bucket_name
        root = (config.root_path or "").strip("/")
        if path.startswith("/"):
            path = path[1:]
        if root and path:
            return f"{bucket}/{root}/{path}"
        elif root:
            return f"{bucket}/{root}"
        elif path:
            return f"{bucket}/{path}"
        return bucket

    def _adapt_entry_path(self, filesystem_path: str, config: GoogleCloudStorageFileSourceConfiguration) -> str:
        """Remove the GCS bucket name and root_path from the filesystem path."""
        if config.bucket_name:
            bucket = config.bucket_name
            root = (config.root_path or "").strip("/")
            full_prefix = f"{bucket}/{root}" if root else bucket
            if filesystem_path == full_prefix:
                return "/"
            return "/" + filesystem_path.removeprefix(f"{full_prefix}/")
        return "/" + filesystem_path


[docs]
    def score_url_match(self, url: str):
        bucket_name = self.template_config.bucket_name
        # For security, we need to ensure that a partial match doesn't work
        if bucket_name and (url.startswith(f"gs://{bucket_name}/") or url == f"gs://{bucket_name}"):
            return len(f"gs://{bucket_name}")
        elif bucket_name and (url.startswith(f"gcs://{bucket_name}/") or url == f"gcs://{bucket_name}"):
            return len(f"gcs://{bucket_name}")
        else:
            return super().score_url_match(url)




__all__ = ("GoogleCloudStorageFilesSource",)