Skip to content

sftp

Strategy class for sftp/ftp

AnyFtpUrl = Annotated[AnyUrl, UrlConstraints(allowed_schemes=['ftp', 'sftp'])] module-attribute

SFTPConfig

Bases: AttrDict

(S)FTP-specific Configuration Data Model.

Source code in oteapi/strategies/download/sftp.py 
20
21
22
23
24
25
26
27
28
29
class SFTPConfig(AttrDict):
    """(S)FTP-specific Configuration Data Model."""

    datacache_config: DataCacheConfig | None = Field(
        None,
        description=(
            "Configurations for the data cache for storing the downloaded file "
            "content."
        ),
    )

datacache_config = Field(None, description='Configurations for the data cache for storing the downloaded file content.') class-attribute instance-attribute

SFTPContent

Bases: AttrDict

Class for returning values from Download SFTP strategy.

Source code in oteapi/strategies/download/sftp.py 
43
44
45
46
class SFTPContent(AttrDict):
    """Class for returning values from Download SFTP strategy."""

    key: str = Field(..., description="Key to access the data in the cache.")

key = Field(..., description='Key to access the data in the cache.') class-attribute instance-attribute

SFTPResourceConfig

Bases: ResourceConfig

(S)FTP download strategy filter config.

Source code in oteapi/strategies/download/sftp.py 
32
33
34
35
36
37
38
39
40
class SFTPResourceConfig(ResourceConfig):
    """(S)FTP download strategy filter config."""

    downloadUrl: AnyFtpUrl = Field(  # type: ignore[assignment]
        ..., description="The (S)FTP URL, which will be downloaded."
    )
    configuration: SFTPConfig = Field(
        SFTPConfig(), description="(S)FTP download strategy-specific configuration."
    )

configuration = Field(SFTPConfig(), description='(S)FTP download strategy-specific configuration.') class-attribute instance-attribute

downloadUrl = Field(..., description='The (S)FTP URL, which will be downloaded.') class-attribute instance-attribute

SFTPStrategy

Strategy for retrieving data via sftp.

Registers strategies:

  • ("scheme", "ftp")
  • ("scheme", "sftp")
Source code in oteapi/strategies/download/sftp.py 
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
@dataclass
class SFTPStrategy:
    """Strategy for retrieving data via sftp.

    **Registers strategies**:

    - `("scheme", "ftp")`
    - `("scheme", "sftp")`

    """

    download_config: SFTPResourceConfig

    def initialize(self) -> AttrDict:
        """Initialize."""
        return AttrDict()

    def get(self) -> SFTPContent:
        """Download via sftp"""
        url = self.download_config.downloadUrl
        if not url.host or not url.path:
            raise ValueError(
                "Invalid (S)FTP URL (missing host or path): "
                f"host={url.host!r}, path={url.path!r}"
            )

        cache = DataCache(self.download_config.configuration.datacache_config)
        if cache.config.accessKey and cache.config.accessKey in cache:
            key = cache.config.accessKey
        else:
            with paramiko.SSHClient() as client:
                client.set_missing_host_key_policy(
                    paramiko.AutoAddPolicy()
                )  # nosec B507
                client.connect(
                    hostname=url.host,
                    username=url.username,
                    password=url.password,
                    port=url.port or 22,
                )
                # Because of insane locking on Windows, we have to close
                # the downloaded file before adding it to the cache
                with NamedTemporaryFile(prefix="oteapi-sftp-", delete=False) as handle:
                    localpath = Path(handle.name).resolve()
                try:
                    with client.open_sftp() as sftp:
                        sftp.get(url.path, str(localpath))
                    key = cache.add(localpath.read_bytes())
                finally:
                    localpath.unlink()

        return SFTPContent(key=key)

download_config instance-attribute

get()

Download via sftp

Source code in oteapi/strategies/download/sftp.py 
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
def get(self) -> SFTPContent:
    """Download via sftp"""
    url = self.download_config.downloadUrl
    if not url.host or not url.path:
        raise ValueError(
            "Invalid (S)FTP URL (missing host or path): "
            f"host={url.host!r}, path={url.path!r}"
        )

    cache = DataCache(self.download_config.configuration.datacache_config)
    if cache.config.accessKey and cache.config.accessKey in cache:
        key = cache.config.accessKey
    else:
        with paramiko.SSHClient() as client:
            client.set_missing_host_key_policy(
                paramiko.AutoAddPolicy()
            )  # nosec B507
            client.connect(
                hostname=url.host,
                username=url.username,
                password=url.password,
                port=url.port or 22,
            )
            # Because of insane locking on Windows, we have to close
            # the downloaded file before adding it to the cache
            with NamedTemporaryFile(prefix="oteapi-sftp-", delete=False) as handle:
                localpath = Path(handle.name).resolve()
            try:
                with client.open_sftp() as sftp:
                    sftp.get(url.path, str(localpath))
                key = cache.add(localpath.read_bytes())
            finally:
                localpath.unlink()

    return SFTPContent(key=key)

initialize()

Initialize.

Source code in oteapi/strategies/download/sftp.py 
62
63
64
def initialize(self) -> AttrDict:
    """Initialize."""
    return AttrDict()