Source code for galaxy.files.sources.ftp

import urllib.parse

try:
    from fs.ftpfs import FTPFS
except ImportError:
    FTPFS = None  # type: ignore[misc,assignment]

from typing import (
    cast,
    Optional,
    Tuple,
)

from . import (
    FilesSourceOptions,
    FilesSourceProperties,
)
from ._pyfilesystem2 import PyFilesystem2FilesSource


class FTPFilesSourceProperties(FilesSourceProperties, total=False):
    host: str
    port: int
    user: str
    passwd: str


[docs]class FtpFilesSource(PyFilesystem2FilesSource):
    plugin_type = "ftp"
    required_module = FTPFS
    required_package = "fs.ftpfs"

    def _open_fs(self, user_context=None, opts: Optional[FilesSourceOptions] = None):
        props = self._serialization_props(user_context)
        extra_props: FTPFilesSourceProperties = cast(FTPFilesSourceProperties, opts.extra_props or {} if opts else {})
        handle = FTPFS(**{**props, **extra_props})
        return handle

    def _realize_to(
        self, source_path: str, native_path: str, user_context=None, opts: Optional[FilesSourceOptions] = None
    ):
        extra_props: FTPFilesSourceProperties
        if opts and opts.extra_props:
            extra_props = cast(FTPFilesSourceProperties, opts.extra_props)
        else:
            opts = FilesSourceOptions()
            extra_props = {}
        path, opts.extra_props = self._get_props_and_rel_path(extra_props, source_path)
        super()._realize_to(path, native_path, user_context=user_context, opts=opts)

    def _write_from(
        self, target_path: str, native_path: str, user_context=None, opts: Optional[FilesSourceOptions] = None
    ):
        extra_props: FTPFilesSourceProperties
        if opts and opts.extra_props:
            extra_props = cast(FTPFilesSourceProperties, opts.extra_props)
        else:
            opts = FilesSourceOptions()
            extra_props = {}
        path, opts.extra_props = self._get_props_and_rel_path(extra_props, target_path)
        super()._write_from(path, native_path, user_context=user_context, opts=opts)

    def _get_props_and_rel_path(
        self, extra_props: FTPFilesSourceProperties, url: str
    ) -> Tuple[str, FTPFilesSourceProperties]:
        host = self._props.get("host")
        port = self._props.get("port")
        user = self._props.get("user")
        passwd = self._props.get("passwd")
        rel_path = url
        if url.startswith(f"ftp://{host or ''}"):
            props = self._extract_url_props(url)
            extra_props["host"] = host or props["host"]
            extra_props["port"] = port or props["port"]
            extra_props["user"] = user or props["user"]
            extra_props["passwd"] = passwd or props["passwd"]
            rel_path = props["path"] or url
        return rel_path, extra_props

    def _extract_url_props(self, url: str):
        result = urllib.parse.urlparse(url)
        return {
            "host": result.hostname,
            "port": result.port or 21,
            "user": result.username,
            "passwd": result.password,
            "path": result.path,
        }

[docs]    def score_url_match(self, url: str):
        host = self._props.get("host")
        port = self._props.get("port")
        if host and port and url.startswith(f"ftp://{host}:{port}"):
            return len(f"ftp://{host}:{port}")
        # For security, we need to ensure that a partial match doesn't work e.g. ftp://{host}something/myfiles
        elif host and (url.startswith(f"ftp://{host}/") or url == f"ftp://{host}"):
            return len(f"ftp://{host}")
        elif not host and url.startswith("ftp://"):
            return len("ftp://")
        else:
            return super().score_url_match(url)


__all__ = ("FtpFilesSource",)