Source code for galaxy.files.sources.ssh

try:
    from fs.sshfs.sshfs import SSHFS
except ImportError:
    SSHFS = None

from typing import (
    Optional,
    Union,
)

from galaxy.files.models import (
    BaseFileSourceConfiguration,
    BaseFileSourceTemplateConfiguration,
    FilesSourceRuntimeContext,
)
from galaxy.util.config_templates import TemplateExpansion
from ._pyfilesystem2 import PyFilesystem2FilesSource


class SshFileSourceTemplateConfiguration(BaseFileSourceTemplateConfiguration):
    host: Union[str, TemplateExpansion]
    user: Optional[Union[str, TemplateExpansion]] = None
    passwd: Optional[Union[str, TemplateExpansion]] = None
    pkey: Optional[Union[str, TemplateExpansion]] = None
    timeout: Union[int, TemplateExpansion] = 10
    port: Union[int, TemplateExpansion] = 22
    compress: Union[bool, TemplateExpansion] = False
    config_path: Union[str, TemplateExpansion] = "~/.ssh/config"
    path: Union[str, TemplateExpansion]


class SshFileSourceConfiguration(BaseFileSourceConfiguration):
    host: str
    user: Optional[str] = None
    passwd: Optional[str] = None
    pkey: Optional[str] = None
    timeout: int = 10
    port: int = 22
    compress: bool = False
    config_path: str = "~/.ssh/config"
    path: str



[docs]
class SshFilesSource(PyFilesystem2FilesSource[SshFileSourceTemplateConfiguration, SshFileSourceConfiguration]):
    plugin_type = "ssh"
    required_module = SSHFS
    required_package = "fs.sshfs"

    template_config_class = SshFileSourceTemplateConfiguration
    resolved_config_class = SshFileSourceConfiguration

    def _open_fs(self, context: FilesSourceRuntimeContext[SshFileSourceConfiguration]):
        if SSHFS is None:
            raise self.required_package_exception

        config = context.config
        handle = SSHFS(
            host=config.host,
            user=config.user,
            passwd=config.passwd,
            pkey=config.pkey,
            port=config.port,
            timeout=config.timeout,
            compress=config.compress,
            config_path=config.config_path,
        )
        if config.path:
            return handle.opendir(config.path)
        return handle



__all__ = ("SshFilesSource",)