Skip to content

gportal.sftp

download(target, local_dir='.', username=None, password=None)

Downloads files to a local directory via SFTP.

Parameters:

Name Type Description Default
target Union[str, Product, Iterable[Union[str, Product]]]

Remote path, Product object, or a list of them.

 required
local_dir str

Local directory to download to.

 '.'
username Optional[str]

G-Portal username. If not provided, the value of gportal.username is used.

 None
password Optional[str]

G-Portal password. If not provided, the value of gportal.password is used.

 None

Returns:

Type Description
Union[str, list[str]]

A local path of the downloaded file when target is a single path or Product object.
Union[str, list[str]]

A list of local paths of the downloaded files when target is a list of paths or Product objects.
Source code in gportal/sftp.py 
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
def download(
    target: Union[str, Product, Iterable[Union[str, Product]]],
    local_dir: str = ".",
    username: Optional[str] = None,
    password: Optional[str] = None,
) -> Union[str, list[str]]:
    """Downloads files to a local directory via SFTP.

    Args:
        target: Remote path, Product object, or a list of them.
        local_dir: Local directory to download to.
        username: G-Portal username. If not provided, the value of `gportal.username` is used.
        password: G-Portal password. If not provided, the value of `gportal.password` is used.

    Returns:
        A local path of the downloaded file when `target` is a single path or Product object.
        A list of local paths of the downloaded files when `target` is a list of paths or Product objects.
    """
    with SFTP.connect(username, password) as sftp:
        return sftp.download(target, local_dir)

SFTP

Wrapper of the G-Portal SFTP interface.

Attributes:

Name Type Description
client SFTPClient

An instance of paramiko.SFTPClient.
Source code in gportal/sftp.py 
 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
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
class SFTP:
    """Wrapper of the G-Portal SFTP interface.

    Attributes:
        client: An instance of [`paramiko.SFTPClient`][paramiko.sftp_client.SFTPClient].
    """

    HOST = "ftp.gportal.jaxa.jp"
    PORT = 2051

    def __init__(self, sftp_client: SFTPClient):
        self.client: SFTPClient = sftp_client

    def __enter__(self):
        return self

    def __exit__(self, exc_type, exc_value, traceback):
        self.close()

    @classmethod
    def connect(cls, username: Optional[str] = None, password: Optional[str] = None) -> "SFTP":
        """Opens an SFTP session with G-Portal.

        Args:
            username: G-Portal username. If not provided, the value of `gportal.username` is used.
            password: G-Portal password. If not provided, the value of `gportal.password` is used.

        Returns:
            An instance of [`SFTP`][gportal.sftp.SFTP].
        """
        username = username or gportal.username
        password = password or gportal.password

        if username is None or password is None:
            raise ValueError("username and password are required")

        transport = Transport((cls.HOST, cls.PORT))
        transport.connect(username=username, password=password)

        sftp_client = transport.open_sftp_client()
        if sftp_client is None:  # will never happen
            raise RuntimeError("Failed to open SFTP session")

        return cls(sftp_client)

    def close(self) -> None:
        """Closes the SFTP session."""
        self.client.close()

    def listdir(self, path: str = "/", /, filter_pattern: Optional[str] = None, fullpath: bool = False) -> list[str]:
        """Returns a list containing the names of the entries in the given path.

        Wraps [`paramiko.SFTPClient.listdir`][paramiko.sftp_client.SFTPClient.listdir].

        Args:
            path: Remote path to list. It must be absolute.
            filter_pattern: Regular expression to filter the entries.
            fullpath: If `True`, the returned list contains full paths of the entries.

        Returns:
            A list containing the names of the entries.
        """
        self._reset_cwd()
        entries = self.client.listdir(path)

        if filter_pattern:
            entries = [entry for entry in entries if re.search(filter_pattern, entry)]

        if fullpath:
            return [os.path.join(path, entry) for entry in entries]
        else:
            return entries

    @overload
    def download(self, target: Union[str, Product], local_dir: str) -> str:
        ...

    @overload
    def download(self, target: Iterable[Union[str, Product]], local_dir: str) -> list[str]:
        ...

    def download(
        self, target: Union[str, Product, Iterable[Union[str, Product]]], local_dir: str
    ) -> Union[str, list[str]]:
        """Downloads files to a local directory.

        Args:
            target: Remote path, Product object, or a list of them.
            local_dir: Local directory to download to.

        Returns:
            A local path of the downloaded file when `target` is a single path or Product object.
            A list of local paths of the downloaded files when `target` is a list of paths or Product objects.
        """
        self._reset_cwd()

        if isinstance(target, Iterable) and not isinstance(target, str):
            return [self._download_single(t, local_dir) for t in target]
        else:
            return self._download_single(target, local_dir)

    def _download_single(self, target: Union[str, Product], local_dir: str) -> str:
        if isinstance(target, Product):
            if target.data_path is None:
                raise ValueError(f"Product {target.id} has no URL to download")

            target = target.data_path

        elif target.startswith("https://gportal.jaxa.jp/download/"):
            target = target.replace("https://gportal.jaxa.jp/download/", "", 1)

        local_path = os.path.join(local_dir, os.path.basename(target))
        self.client.get(target, local_path)
        return local_path

    def _reset_cwd(self) -> None:
        self.client.chdir()

connect(username=None, password=None) classmethod

Opens an SFTP session with G-Portal.

Parameters:

Name Type Description Default
username Optional[str]

G-Portal username. If not provided, the value of gportal.username is used.

 None
password Optional[str]

G-Portal password. If not provided, the value of gportal.password is used.

 None

Returns:

Type Description
SFTP

An instance of SFTP.
Source code in gportal/sftp.py 
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
@classmethod
def connect(cls, username: Optional[str] = None, password: Optional[str] = None) -> "SFTP":
    """Opens an SFTP session with G-Portal.

    Args:
        username: G-Portal username. If not provided, the value of `gportal.username` is used.
        password: G-Portal password. If not provided, the value of `gportal.password` is used.

    Returns:
        An instance of [`SFTP`][gportal.sftp.SFTP].
    """
    username = username or gportal.username
    password = password or gportal.password

    if username is None or password is None:
        raise ValueError("username and password are required")

    transport = Transport((cls.HOST, cls.PORT))
    transport.connect(username=username, password=password)

    sftp_client = transport.open_sftp_client()
    if sftp_client is None:  # will never happen
        raise RuntimeError("Failed to open SFTP session")

    return cls(sftp_client)

close()

Closes the SFTP session.

Source code in gportal/sftp.py 
101
102
103
def close(self) -> None:
    """Closes the SFTP session."""
    self.client.close()

listdir(path='/', /, filter_pattern=None, fullpath=False)

Returns a list containing the names of the entries in the given path.

Wraps paramiko.SFTPClient.listdir.

Parameters:

Name Type Description Default
path str

Remote path to list. It must be absolute.

 '/'
filter_pattern Optional[str]

Regular expression to filter the entries.

 None
fullpath bool

If True, the returned list contains full paths of the entries.

 False

Returns:

Type Description
list[str]

A list containing the names of the entries.
Source code in gportal/sftp.py 
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
def listdir(self, path: str = "/", /, filter_pattern: Optional[str] = None, fullpath: bool = False) -> list[str]:
    """Returns a list containing the names of the entries in the given path.

    Wraps [`paramiko.SFTPClient.listdir`][paramiko.sftp_client.SFTPClient.listdir].

    Args:
        path: Remote path to list. It must be absolute.
        filter_pattern: Regular expression to filter the entries.
        fullpath: If `True`, the returned list contains full paths of the entries.

    Returns:
        A list containing the names of the entries.
    """
    self._reset_cwd()
    entries = self.client.listdir(path)

    if filter_pattern:
        entries = [entry for entry in entries if re.search(filter_pattern, entry)]

    if fullpath:
        return [os.path.join(path, entry) for entry in entries]
    else:
        return entries

download(target, local_dir)

Downloads files to a local directory.

Parameters:

Name Type Description Default
target Union[str, Product, Iterable[Union[str, Product]]]

Remote path, Product object, or a list of them.

 required
local_dir str

Local directory to download to.

 required

Returns:

Type Description
Union[str, list[str]]

A local path of the downloaded file when target is a single path or Product object.
Union[str, list[str]]

A list of local paths of the downloaded files when target is a list of paths or Product objects.
Source code in gportal/sftp.py 
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
def download(
    self, target: Union[str, Product, Iterable[Union[str, Product]]], local_dir: str
) -> Union[str, list[str]]:
    """Downloads files to a local directory.

    Args:
        target: Remote path, Product object, or a list of them.
        local_dir: Local directory to download to.

    Returns:
        A local path of the downloaded file when `target` is a single path or Product object.
        A list of local paths of the downloaded files when `target` is a list of paths or Product objects.
    """
    self._reset_cwd()

    if isinstance(target, Iterable) and not isinstance(target, str):
        return [self._download_single(t, local_dir) for t in target]
    else:
        return self._download_single(target, local_dir)