Source code for galaxy.files.sources.dropbox

try:
    from fs.dropboxfs import DropboxFS
except ImportError:
    DropboxFS = None

from typing import (
    Optional,
    Union,
)

from galaxy.exceptions import (
    AuthenticationRequired,
    MessageException,
)
from . import (
    FilesSourceOptions,
    FilesSourceProperties,
)
from ._pyfilesystem2 import PyFilesystem2FilesSource



[docs]
class DropboxFilesSource(PyFilesystem2FilesSource):
    plugin_type = "dropbox"
    required_module = DropboxFS
    required_package = "fs.dropboxfs"

    def _open_fs(self, user_context=None, opts: Optional[FilesSourceOptions] = None):
        props = self._serialization_props(user_context)
        extra_props: Union[FilesSourceProperties, dict] = opts.extra_props or {} if opts else {}
        # accessToken has been renamed to access_token in fs.dropboxfs 1.0
        if "accessToken" in props:
            props["access_token"] = props.pop("accessToken")
        if "oauth2_access_token" in props:
            props["access_token"] = props.pop("oauth2_access_token")

        try:
            handle = DropboxFS(**{**props, **extra_props})
            return handle
        except Exception as e:
            # This plugin might raise dropbox.dropbox_client.BadInputException
            # which is not a subclass of fs.errors.FSError
            if "OAuth2" in str(e):
                raise AuthenticationRequired(
                    f"Permission Denied. Reason: {e}. Please check your credentials in your preferences for {self.label}."
                )
            raise MessageException(f"Error connecting to Dropbox. Reason: {e}")



__all__ = ("DropboxFilesSource",)