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: Optional[DataCacheConfig] = Field(
        None,
        description=(
            "Configurations for the data cache for storing the downloaded file "
            "content."
        ),
    )

datacache_config: Optional[DataCacheConfig] = 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: str = 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: SFTPConfig = Field(SFTPConfig(), description='(S)FTP download strategy-specific configuration.') class-attribute instance-attribute

downloadUrl: AnyFtpUrl = 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
@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"""
        cache = DataCache(self.download_config.configuration.datacache_config)
        if cache.config.accessKey and cache.config.accessKey in cache:
            key = cache.config.accessKey
        else:
            # Setup connection options
            cnopts = pysftp.CnOpts()
            cnopts.hostkeys = None

            # open connection and store data locally
            with pysftp.Connection(
                host=self.download_config.downloadUrl.host,
                username=self.download_config.downloadUrl.username,
                password=self.download_config.downloadUrl.password,
                port=self.download_config.downloadUrl.port,
                cnopts=cnopts,
            ) as sftp:
                # 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:
                    sftp.get(self.download_config.downloadUrl.path, localpath=localpath)
                    key = cache.add(localpath.read_bytes())
                finally:
                    localpath.unlink()

        return SFTPContent(key=key)

download_config: SFTPResourceConfig 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
def get(self) -> SFTPContent:
    """Download via sftp"""
    cache = DataCache(self.download_config.configuration.datacache_config)
    if cache.config.accessKey and cache.config.accessKey in cache:
        key = cache.config.accessKey
    else:
        # Setup connection options
        cnopts = pysftp.CnOpts()
        cnopts.hostkeys = None

        # open connection and store data locally
        with pysftp.Connection(
            host=self.download_config.downloadUrl.host,
            username=self.download_config.downloadUrl.username,
            password=self.download_config.downloadUrl.password,
            port=self.download_config.downloadUrl.port,
            cnopts=cnopts,
        ) as sftp:
            # 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:
                sftp.get(self.download_config.downloadUrl.path, localpath=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()