tenseleyflow/wulftp / 905b3e4

Browse files

refactor

Authored by espadonne
SHA
905b3e438d0bcfb68d1344e46aa4a8acfb308e60
Parents
7ea1b01
Tree
b34e29f

3 changed files

StatusFile+-
R wulftp/__init__.py → src/wulftp/__init__.py 0 0
A src/wulftp/sftp_backend.py 51 0
R wulftp/wulftp.py → src/wulftp/wulftp.py 0 0
wulftp/__init__.py → src/wulftp/__init__.pyrenamed (100% similarity)
src/wulftp/sftp_backend.pyadded
@@ -0,0 +1,51 @@
1
+import paramiko
2
+
3
+class SFTPBackend:
4
+    """
5
+    simple wrapper for SFTP operations using Paramiko.
6
+    """
7
+
8
+    def __init__(self):
9
+        self.ssh = None
10
+        self.sftp = None
11
+
12
+    def connect(self, host: str, port: int, username: str, key_path: str, passphrase: str | None = None):
13
+        """
14
+        establish an SSH connection and open an SFTP session.
15
+        Raises Paramiko exceptions on failure.
16
+        """
17
+        self.ssh = paramiko.SSHClient()
18
+        self.ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
19
+        if passphrase:
20
+            pkey = paramiko.RSAKey.from_private_key_file(key_path, password=passphrase)
21
+            self.ssh.connect(hostname=host, port=port, username=username, pkey=pkey)
22
+        else:
23
+            self.ssh.connect(hostname=host, port=port, username=username, key_filename=key_path)
24
+        self.sftp = self.ssh.open_sftp()
25
+
26
+    def listdir(self, path: str = '.') -> list[str]:
27
+        """
28
+        return the list of entries in the remote directory.
29
+        """
30
+        return self.sftp.listdir(path)
31
+
32
+    def upload(self, local_path: str, remote_path: str) -> None:
33
+        """
34
+        upload a local file to the specified remote path.
35
+        """
36
+        self.sftp.put(local_path, remote_path)
37
+
38
+    def download(self, remote_path: str, local_path: str) -> None:
39
+        """
40
+        download a remote file to the specified local path.
41
+        """
42
+        self.sftp.get(remote_path, local_path)
43
+
44
+    def close(self) -> None:
45
+        """
46
+        close the SFTP session and underlying SSH connection.
47
+        """
48
+        if self.sftp:
49
+            self.sftp.close()
50
+        if self.ssh:
51
+            self.ssh.close()
wulftp/wulftp.py → src/wulftp/wulftp.pyrenamed (92% similarity)
@@ -1,4 +1,3 @@
1
-
21
 import os
32
 import sys
43
 import threading
@@ -13,6 +12,8 @@ from PyQt6.QtWidgets import (
1312
     QLineEdit, QToolButton, QPushButton, QListWidget, QFileDialog, QMessageBox, QFormLayout,
1413
 )
1514
15
+from .sftp_backend import SFTPBackend
16
+
1617
1718
 # fetch default host/port from .env
1819
 load_dotenv()
@@ -23,7 +24,7 @@ DEFAULT_PORT = int(os.getenv('WULFTP_PORT', '22'))
2324
 class WulFTPClient(QMainWindow):
2425
     def __init__(self):
2526
         super().__init__()
26
-        self.ssh = None
27
+        self.backend = SFTPBackend()
2728
         self.sftp = None
2829
         self.init_ui()
2930
@@ -154,7 +155,7 @@ class WulFTPClient(QMainWindow):
154155
155156
             def task(lp=local_path, rf=remote_file):
156157
                 try:
157
-                    self.sftp.put(lp, rf)
158
+                    self.backend.upload(lp, rf)
158159
                     self.refresh_remote()
159160
                 except Exception as e:
160161
                     self.show_error(str(e))
@@ -183,14 +184,8 @@ class WulFTPClient(QMainWindow):
183184
         passphrase = self.passphrase_input.text() or None
184185
185186
         try:
186
-            self.ssh = paramiko.SSHClient()
187
-            self.ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
188
-            if passphrase:
189
-                pkey = paramiko.RSAKey.from_private_key_file(key_path, password=passphrase)
190
-                self.ssh.connect(hostname=host, port=port, username=user, pkey=pkey)
191
-            else:
192
-                self.ssh.connect(hostname=host, port=port, username=user, key_filename=key_path)
193
-            self.sftp = self.ssh.open_sftp()
187
+            self.backend.connect(host, port, user, key_path, passphrase)
188
+            self.sftp = self.backend.sftp
194189
195190
             self.update_status_icon(True)
196191
             self.browse_file_btn.setEnabled(True)
@@ -208,7 +203,7 @@ class WulFTPClient(QMainWindow):
208203
         try:
209204
             self.remote_path = '.'
210205
             self.remote_list.clear()
211
-            for f in self.sftp.listdir(self.remote_path):
206
+            for f in self.backend.listdir(self.remote_path):
212207
                 self.remote_list.addItem(f)
213208
         except Exception as e:
214209
             self.show_error(str(e))
@@ -224,7 +219,7 @@ class WulFTPClient(QMainWindow):
224219
225220
         def task():
226221
             try:
227
-                self.sftp.put(path, remote_file)
222
+                self.backend.upload(path, remote_file)
228223
                 self.refresh_remote()
229224
             except Exception as e:
230225
                 self.show_error(str(e))