snoopy/pyssh-client
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: rendre les operations SFTP asynchrones

Browse Source 
RemoteFilePanel appelait sftp.listdir_attr/mkdir/rename/remove/chmod/get
directement depuis le thread GUI : un dossier distant lent ou une latence
reseau gelait toute l'interface. Seuls upload/download passaient deja par un
thread dedie (TransferWorker), et celui-ci reutilisait le meme SFTPClient que
la navigation - paramiko.SFTPClient n'est pas concu pour etre pilote par
plusieurs threads a la fois.

- app/sftp/worker.py : SftpWorker, un thread proprietaire exclusif d'un canal
  SFTP, avec une file de jobs (callables) et les resultats renvoyes par signal
  Qt (marshalling automatique vers le thread GUI).
- RemoteFilePanel soumet desormais toutes ses operations via ce worker avec
  callbacks succes/erreur ; un compteur de generation ignore les reponses de
  navigations devenues obsoletes (double-clic rapide, edition puis navigation
  ailleurs pendant le renvoi).
- TransferWorker ouvre son propre canal SFTP (`sftp_factory`) au lieu de
  partager celui de la navigation : un gros transfert ne bloque plus le
  listage de dossiers. SFTPPanel(sftp, transfer_sftp_factory=...) permet
  d'injecter ce second canal (paramiko.SFTPClient.from_transport) ; a defaut,
  le canal de navigation est reutilise tel quel (cas des tests).
- SessionTabWidget/SFTPTabWidget ouvrent ce second canal et appellent
  SFTPPanel.shutdown() a la fermeture de l'onglet (arrete proprement les
  threads avant destruction des widgets).

tests/test_sftp.py adapte pour pomper la boucle d'evenements Qt apres les
operations desormais asynchrones (meme couverture, timing different).

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
Florian
2026-09-05 01:42:46 +02:00
parent 47005b1e26
commit fd848c2025
4 changed files with 326 additions and 108 deletions
Download Patch File Download Diff File Expand all files Collapse all files

0
app/sftp/__init__.py Normal file
View File

59
app/sftp/worker.py Normal file
View File

@@ -0,0 +1,59 @@
"""Execution asynchrone des operations SFTP.
paramiko.SFTPClient n'est pas concu pour etre pilote depuis plusieurs threads a
la fois : ce module fait en sorte qu'un seul thread possede jamais un client SFTP
donne. Toute operation (listage, mkdir, rename, remove, chmod, stat, ...) est
soumise a la file de ce thread sous forme de callable ; le resultat revient sur
le thread appelant (GUI) via un signal Qt, qui les mets automatiquement en file
d'attente quand emetteur et recepteur vivent dans des threads differents.
"""
from __future__ import annotations
import queue
import threading
from dataclasses import dataclass
from typing import Any, Callable
from PySide6.QtCore import QThread, Signal
@dataclass
class _Job:
request_id: int
fn: Callable[[], Any]
class SftpWorker(QThread):
"""Un worker = un canal SFTP, jamais partage avec un autre thread."""
result_ready = Signal(int, object, object) # request_id, resultat, exception|None
def __init__(self, parent=None):
super().__init__(parent)
self._queue: queue.Queue[_Job | None] = queue.Queue()
self._next_id = 0
self._lock = threading.Lock()
def submit(self, fn: Callable[[], Any]) -> int:
"""Met `fn` en file pour execution sur le thread du worker. Renvoie un
identifiant a rapprocher du signal `result_ready`."""
with self._lock:
self._next_id += 1
request_id = self._next_id
self._queue.put(_Job(request_id, fn))
return request_id
def run(self) -> None:
while True:
job = self._queue.get()
if job is None: # sentinelle posee par stop()
break
try:
result = job.fn()
self.result_ready.emit(job.request_id, result, None)
except Exception as exc: # noqa: BLE001 - remonte a l'appelant via le signal
self.result_ready.emit(job.request_id, None, exc)
def stop(self) -> None:
"""Arrete le worker une fois les jobs deja en file executes."""
self._queue.put(None)

230
app/sftp_panel.py
View File

@@ -19,6 +19,8 @@ from PySide6.QtWidgets import (
QAbstractItemView, QMenu, QHeaderView,
)
from app.sftp.worker import SftpWorker
MIME_TYPE = "application/x-pyssh-files"
NAME_ROLE = Qt.UserRole
@@ -56,16 +58,25 @@ class TransferJob:
class TransferWorker(QThread):
"""Execute une file de transferts sequentiellement, dans son propre thread."""
"""Execute une file de transferts sequentiellement, dans son propre thread.
Ouvre son propre canal SFTP (via `sftp_factory`) plutot que de reutiliser celui
de la navigation : un gros transfert ne doit jamais faire attendre un listage
de dossier, et paramiko.SFTPClient n'est pas concu pour etre partage entre
threads. Quand `close_when_done` est faux (aucun canal dedie disponible), le
canal fourni est reutilise tel quel et n'est pas ferme en fin de transfert.
"""
file_started = Signal(str, int, int) # nom, index, total
file_progress = Signal(int, int) # octets transferes, taille totale
finished_all = Signal(int, list) # nb reussis, liste d'erreurs
def __init__(self, sftp: paramiko.SFTPClient, jobs: list[TransferJob]):
def __init__(self, sftp_factory, jobs: list[TransferJob], close_when_done: bool = False):
super().__init__()
self.sftp = sftp
self.sftp_factory = sftp_factory
self.sftp: paramiko.SFTPClient | None = None
self.jobs = jobs
self.close_when_done = close_when_done
self._cancel = False
self._done = 0
self._errors: list[str] = []
@@ -74,6 +85,12 @@ class TransferWorker(QThread):
self._cancel = True
def run(self) -> None:
try:
self.sftp = self.sftp_factory()
except Exception as exc: # noqa: BLE001
self.finished_all.emit(0, [f"Impossible d'ouvrir le canal SFTP : {exc}"])
return
try:
total = len(self.jobs)
for index, job in enumerate(self.jobs, start=1):
if self._cancel:
@@ -95,6 +112,12 @@ class TransferWorker(QThread):
except Exception as exc: # noqa: BLE001
self._errors.append(f"{label} : {exc}")
self.finished_all.emit(self._done, self._errors)
finally:
if self.close_when_done:
try:
self.sftp.close()
except Exception:
pass
def _cb(self, transferred: int, total: int) -> None:
self.file_progress.emit(transferred, total)
@@ -456,18 +479,32 @@ class LocalFilePanel(FilePanel):
class RemoteFilePanel(FilePanel):
def __init__(self, sftp: paramiko.SFTPClient, parent=None):
"""Toutes les operations reseau passent par `worker` (un `SftpWorker` dedie a ce
panneau) : aucun appel a `self.sftp` ne doit avoir lieu directement depuis le
thread GUI, pour ne jamais geler l'interface et pour respecter le fait qu'un
SFTPClient paramiko ne doit etre pilote que par un seul thread a la fois."""
def __init__(self, sftp: paramiko.SFTPClient, worker: SftpWorker, parent=None):
super().__init__("🌐 Distant", "remote", parent)
self.sftp = sftp
self.worker = worker
self.worker.result_ready.connect(self._on_result)
self._callbacks: dict[int, tuple] = {}
self._nav_generation = 0
# fichiers ouverts pour edition : chemin local temporaire -> chemin distant
self._edited: dict[str, str] = {}
self._watcher = QFileSystemWatcher(self)
self._watcher.fileChanged.connect(self._on_edited_file_changed)
try:
home = self.sftp.normalize(".")
except Exception:
home = "/"
self.navigate(home)
self.status.setText("Connexion...")
def on_home_success(home) -> None:
self.navigate(home or "/")
def on_home_error(_exc) -> None:
self.navigate("/")
self._run(lambda: self.sftp.normalize("."), on_home_success, on_home_error)
def join(self, a: str, b: str) -> str:
return posixpath.join(a, b)
@@ -475,17 +512,45 @@ class RemoteFilePanel(FilePanel):
def parent_of(self, path: str) -> str:
return posixpath.dirname(path.rstrip("/")) or "/"
def navigate(self, path: str) -> None:
try:
attrs = self.sftp.listdir_attr(path)
except IOError as exc:
QMessageBox.warning(self, "Erreur SFTP", f"{path} : {exc}")
# ---------- soumission asynchrone ----------
def _run(self, fn, on_success=None, on_error=None) -> None:
request_id = self.worker.submit(fn)
self._callbacks[request_id] = (on_success, on_error)
def _on_result(self, request_id: int, result, error: Exception | None) -> None:
callbacks = self._callbacks.pop(request_id, None)
if callbacks is None:
return
on_success, on_error = callbacks
if error is not None:
if on_error:
on_error(error)
else:
QMessageBox.critical(self, "Erreur SFTP", str(error))
elif on_success:
on_success(result)
def navigate(self, path: str) -> None:
self._nav_generation += 1
generation = self._nav_generation
self.status.setText("Chargement...")
def on_success(attrs) -> None:
if generation != self._nav_generation:
return # une navigation plus recente a ete lancee entre-temps
self.current_path = path
self.path_edit.setText(path)
self._populate(attrs)
self.path_changed.emit(path)
def on_error(exc) -> None:
if generation != self._nav_generation:
return
self.status.setText("")
QMessageBox.warning(self, "Erreur SFTP", f"{path} : {exc}")
self._run(lambda: self.sftp.listdir_attr(path), on_success, on_error)
def refresh(self) -> None:
if self.current_path:
self.navigate(self.current_path)
@@ -507,12 +572,13 @@ class RemoteFilePanel(FilePanel):
name, ok = QInputDialog.getText(self, "Nouveau dossier", "Nom du dossier :")
if not ok or not name:
return
try:
self.sftp.mkdir(self.join(self.current_path, name))
self.refresh()
except IOError as exc:
path = self.join(self.current_path, name)
def on_error(exc) -> None:
QMessageBox.critical(self, "Erreur", str(exc))
self._run(lambda: self.sftp.mkdir(path), lambda _r: self.refresh(), on_error)
def rename_selected(self) -> None:
entries = self.selected_entries()
if not entries:
@@ -521,12 +587,14 @@ class RemoteFilePanel(FilePanel):
new_name, ok = QInputDialog.getText(self, "Renommer", "Nouveau nom :", text=old_name)
if not ok or not new_name or new_name == old_name:
return
try:
self.sftp.rename(self.join(self.current_path, old_name), self.join(self.current_path, new_name))
self.refresh()
except IOError as exc:
old_path = self.join(self.current_path, old_name)
new_path = self.join(self.current_path, new_name)
def on_error(exc) -> None:
QMessageBox.critical(self, "Erreur", str(exc))
self._run(lambda: self.sftp.rename(old_path, new_path), lambda _r: self.refresh(), on_error)
def delete_selected(self) -> None:
entries = self.selected_entries()
if not entries:
@@ -536,18 +604,31 @@ class RemoteFilePanel(FilePanel):
f"Supprimer definitivement {len(entries)} element(s) sur le serveur ?",
) != QMessageBox.Yes:
return
current_path = self.current_path
def do_delete() -> list[str]:
errors = []
for name, is_dir in entries:
path = self.join(self.current_path, name)
path = self.join(current_path, name)
try:
if is_dir:
self._rmdir_recursive(path)
else:
self.sftp.remove(path)
except IOError as exc:
QMessageBox.critical(self, "Erreur", f"{name} : {exc}")
errors.append(f"{name} : {exc}")
return errors
def on_success(errors: list[str]) -> None:
if errors:
QMessageBox.critical(self, "Erreur", "\n".join(errors))
self.refresh()
self._run(do_delete, on_success)
def _rmdir_recursive(self, path: str) -> None:
"""Appele uniquement depuis le thread du worker (jamais directement sur le
thread GUI) : parcourt et supprime un dossier distant recursivement."""
for attr in self.sftp.listdir_attr(path):
sub = posixpath.join(path, attr.filename)
if stat.S_ISDIR(attr.st_mode):
@@ -570,12 +651,10 @@ class RemoteFilePanel(FilePanel):
local_dir = Path(tempfile.gettempdir()) / "pyssh-edit"
local_dir.mkdir(parents=True, exist_ok=True)
local_path = local_dir / name
try:
self.sftp.get(remote_path, str(local_path))
except (IOError, OSError) as exc:
QMessageBox.critical(self, "Erreur", f"Telechargement impossible : {exc}")
return
self.status.setText(f"Telechargement de {name}...")
def on_success(_result) -> None:
self._edited[str(local_path)] = remote_path
self._watcher.addPath(str(local_path))
self.status.setText(f"Edition de {name} — enregistrez pour renvoyer sur le serveur")
@@ -588,34 +667,62 @@ class RemoteFilePanel(FilePanel):
f"Ouvrez-le manuellement ; il sera renvoye a chaque enregistrement.\n({exc})",
)
def on_error(exc) -> None:
self.status.setText("")
QMessageBox.critical(self, "Erreur", f"Telechargement impossible : {exc}")
self._run(lambda: self.sftp.get(remote_path, str(local_path)), on_success, on_error)
def _on_edited_file_changed(self, local_path: str) -> None:
remote_path = self._edited.get(local_path)
if not remote_path:
return
try:
self.sftp.put(local_path, remote_path)
except (IOError, OSError) as exc:
def on_error(exc) -> None:
self.status.setText(f"Echec du renvoi : {exc}")
return
# Beaucoup d'editeurs enregistrent en remplacant le fichier : le chemin surveille
# disparait alors du watcher, il faut le reinscrire pour les enregistrements suivants.
def on_success(_result) -> None:
# Beaucoup d'editeurs enregistrent en remplacant le fichier : le chemin
# surveille disparait alors du watcher, il faut le reinscrire pour les
# enregistrements suivants.
if local_path not in self._watcher.files() and Path(local_path).exists():
self._watcher.addPath(local_path)
self.refresh()
# apres refresh() : sinon le message serait ecrase par le compteur d'elements
self._refresh_after_edit(remote_path)
self._run(lambda: self.sftp.put(local_path, remote_path), on_success, on_error)
def _refresh_after_edit(self, remote_path: str) -> None:
"""Comme refresh(), mais affiche un message de confirmation specifique une
fois le listage revenu, au lieu du compteur d'elements habituel."""
self._nav_generation += 1
generation = self._nav_generation
path = self.current_path
def on_success(attrs) -> None:
if generation != self._nav_generation:
return
self._populate(attrs)
self.status.setText(f"↑ {posixpath.basename(remote_path)} renvoye sur le serveur")
def on_error(exc) -> None:
if generation != self._nav_generation:
return
QMessageBox.warning(self, "Erreur SFTP", f"{path} : {exc}")
self._run(lambda: self.sftp.listdir_attr(path), on_success, on_error)
def change_permissions(self) -> None:
entries = self.selected_entries()
if not entries:
return
name = entries[0][0]
path = self.join(self.current_path, name)
try:
current = stat.S_IMODE(self.sftp.stat(path).st_mode)
except IOError as exc:
def on_stat_error(exc) -> None:
QMessageBox.critical(self, "Erreur", str(exc))
return
def on_stat_success(attr) -> None:
current = stat.S_IMODE(attr.st_mode)
text, ok = QInputDialog.getText(
self, "Permissions", f"Droits octaux pour {name} :", text=f"{current:03o}",
)
@@ -626,25 +733,45 @@ class RemoteFilePanel(FilePanel):
except ValueError:
QMessageBox.warning(self, "Valeur invalide", "Entrez une valeur octale, par exemple 644 ou 755.")
return
self._apply_chmod(entries, mode)
self._run(lambda: self.sftp.stat(path), on_stat_success, on_stat_error)
def _apply_chmod(self, entries: list[tuple[str, bool]], mode: int) -> None:
current_path = self.current_path
def do_chmod() -> list[str]:
errors = []
for entry_name, _ in entries:
try:
self.sftp.chmod(self.join(self.current_path, entry_name), mode)
self.sftp.chmod(self.join(current_path, entry_name), mode)
except IOError as exc:
errors.append(f"{entry_name} : {exc}")
return errors
def on_success(errors: list[str]) -> None:
if errors:
QMessageBox.critical(self, "Erreur", "\n".join(errors))
self.refresh()
self._run(do_chmod, on_success)
# ----------------------------------------------------------------------------
# Conteneur double panneau
# ----------------------------------------------------------------------------
class SFTPPanel(QWidget):
def __init__(self, sftp: paramiko.SFTPClient, parent=None):
"""`transfer_sftp_factory`, si fourni, ouvre un second canal SFTP reserve aux
transferts (upload/download), independant du canal de navigation : un gros
transfert ne bloque alors jamais le listage de dossiers. A defaut (ex. client
SFTP de test sans transport reel), le canal de navigation est reutilise tel quel."""
def __init__(self, sftp: paramiko.SFTPClient, transfer_sftp_factory=None, parent=None):
super().__init__(parent)
self.sftp = sftp
self._worker: TransferWorker | None = None
self._dedicated_transfer_channel = transfer_sftp_factory is not None
self._transfer_sftp_factory = transfer_sftp_factory or (lambda: sftp)
layout = QVBoxLayout(self)
layout.setContentsMargins(4, 4, 4, 4)
@@ -662,7 +789,9 @@ class SFTPPanel(QWidget):
splitter = QSplitter(Qt.Horizontal)
self.local = LocalFilePanel()
self.remote = RemoteFilePanel(sftp)
self.remote_worker = SftpWorker(self)
self.remote_worker.start()
self.remote = RemoteFilePanel(sftp, self.remote_worker)
splitter.addWidget(self.local)
splitter.addWidget(self.remote)
splitter.setSizes([500, 500])
@@ -674,6 +803,15 @@ class SFTPPanel(QWidget):
self.local.transfer_here.connect(self._on_transfer_payload)
self.remote.transfer_here.connect(self._on_transfer_payload)
def shutdown(self) -> None:
"""A appeler avant que le panneau ne soit detruit (fermeture de l'onglet) :
arrete proprement le worker de navigation et tout transfert en cours."""
if self._worker and self._worker.isRunning():
self._worker.cancel()
self._worker.wait(2000)
self.remote_worker.stop()
self.remote_worker.wait(2000)
def _on_transfer_payload(self, payload: dict) -> None:
source_side = payload.get("side")
entries = payload.get("entries", [])
@@ -708,7 +846,9 @@ class SFTPPanel(QWidget):
dialog.setAutoClose(False)
dialog.setAutoReset(False)
worker = TransferWorker(self.sftp, jobs)
worker = TransferWorker(
self._transfer_sftp_factory, jobs, close_when_done=self._dedicated_transfer_channel,
)
worker.file_started.connect(
lambda name, i, total: dialog.setLabelText(f"({i}/{total}) {name}")
)

19
tests/test_sftp.py
View File

@@ -70,6 +70,19 @@ class FakeSFTP:
app = QApplication([])
def wait_idle(panel_side, timeout=5.0):
"""Pompe la boucle d'evenements Qt jusqu'a ce que toutes les operations SFTP
soumises par ce panneau (navigate/refresh/etc, desormais asynchrones) soient
revenues."""
deadline = time.time() + timeout
app.processEvents()
while panel_side._callbacks and time.time() < deadline:
app.processEvents()
time.sleep(0.01)
app.processEvents()
remote_root = Path(tempfile.mkdtemp())
local_root = Path(tempfile.mkdtemp())
@@ -85,6 +98,7 @@ local_root = Path(tempfile.mkdtemp())
panel = SFTPPanel(FakeSFTP(remote_root))
panel.local.navigate(str(local_root))
panel.resize(1000, 600)
wait_idle(panel.remote) # laisse le temps a la navigation initiale (asynchrone) d'arriver
print("=== 1. Listing distant avec colonnes ===")
tree = panel.remote.tree
@@ -135,11 +149,13 @@ assert got.exists()
print()
print("=== 5. Renommage distant ===")
panel.remote.navigate("/")
wait_idle(panel.remote)
item = next(panel.remote.tree.topLevelItem(i) for i in range(panel.remote.tree.topLevelItemCount())
if panel.remote.tree.topLevelItem(i).data(0, NAME_ROLE) == "config.yml")
item.setSelected(True)
panel.remote.sftp.rename("/config.yml", "/config_renomme.yml")
panel.remote.refresh()
wait_idle(panel.remote)
names = [panel.remote.tree.topLevelItem(i).data(0, NAME_ROLE)
for i in range(panel.remote.tree.topLevelItemCount())]
print(" entrees apres renommage:", [n for n in names if n])
@@ -155,6 +171,7 @@ print()
print("=== 7. Suppression recursive distante ===")
panel.remote._rmdir_recursive("/dossier_local")
panel.remote.refresh()
wait_idle(panel.remote)
print(" dossier supprime du serveur:", not (remote_root / "dossier_local").exists())
assert not (remote_root / "dossier_local").exists()
@@ -165,5 +182,7 @@ has_parent = any(panel.remote.tree.topLevelItem(i).data(0, PARENT_ROLE)
print(" ligne '..' presente:", has_parent)
assert has_parent
panel.shutdown()
print()
print("TOUS LES TESTS SFTP OK")