快捷導(dǎo)航

Python+Paramiko編寫簡單的SFTP文件上傳下載工具類

更新時(shí)間：2026年01月05日 08:34:21 作者：weixin_46244623

Paramiko 是 Python 中最流行的 SSH2 協(xié)議實(shí)現(xiàn)庫,基于它我們可以輕松實(shí)現(xiàn) SFTP 客戶端功能,本文將分享一個(gè)封裝好的 SftpClient 類,支持文件上傳、下載,感興趣的小伙伴可以了解下

在實(shí)際項(xiàng)目開發(fā)中，我們經(jīng)常需要通過 SFTP 協(xié)議與遠(yuǎn)程服務(wù)器進(jìn)行文件傳輸。Paramiko 是 Python 中最流行的 SSH2 協(xié)議實(shí)現(xiàn)庫，基于它我們可以輕松實(shí)現(xiàn) SFTP 客戶端功能。

本文將分享一個(gè)封裝好的 SftpClient 類，支持文件上傳、下載（帶重試機(jī)制）、上傳進(jìn)度提示以及連接管理，代碼簡潔實(shí)用，可直接拷貝到項(xiàng)目中使用。

環(huán)境準(zhǔn)備

首先確保安裝了 paramiko：

pip install paramiko

完整代碼實(shí)現(xiàn)

from paramiko import Transport, SFTPClient
import time
import errno
import logging

# 配置日志，便于觀察上傳進(jìn)度
logging.basicConfig(format='%(levelname)s : %(message)s',
                    level=logging.INFO)

class SftpClient:
    _connection = None  # 類級別共享的連接對象

    def __init__(self, host, port, username, password):
        self.host = host
        self.port = port
        self.username = username
        self.password = password
        self.create_connection(self.host, self.port,
                               self.username, self.password)

    @classmethod
    def create_connection(cls, host, port, username, password):
        """建立 SFTP 連接"""
        transport = Transport(sock=(host, port))
        transport.connect(username=username, password=password)
        cls._connection = SFTPClient.from_transport(transport)
        logging.info("SFTP 連接成功：{}@{}:{}".format(username, host, port))

    @staticmethod
    def uploading_info(uploaded_file_size, total_file_size):
        """上傳進(jìn)度回調(diào)函數(shù)"""
        percentage = (uploaded_file_size / total_file_size) * 100
        logging.info('已上傳：{} / {} ({:.2f}%)'.format(
            uploaded_file_size, total_file_size, percentage))

    def upload(self, local_path, remote_path):
        """上傳文件，支持進(jìn)度顯示"""
        self._connection.put(localpath=local_path,
                             remotepath=remote_path,
                             callback=self.uploading_info,
                             confirm=True)
        logging.info("文件上傳成功：{} -> {}".format(local_path, remote_path))

    def file_exists(self, remote_path):
        """判斷遠(yuǎn)程文件是否存在"""
        try:
            self._connection.stat(remote_path)
            return True
        except IOError as e:
            if e.errno == errno.ENOENT:
                return False
            raise  # 其他錯(cuò)誤向上拋出

    def download(self, remote_path, local_path, retry=5):
        """
        下載文件
        如果文件不存在，會等待5秒后重試，最多重試5次（適合文件正在生成場景）
        """
        if self.file_exists(remote_path):
            self._connection.get(remote_path, local_path)
            logging.info("文件下載成功：{} -> {}".format(remote_path, local_path))
        elif retry > 0:
            logging.warning("遠(yuǎn)程文件不存在，{}秒后重試（剩余重試次數(shù)：{}）".format(5, retry))
            time.sleep(5)
            self.download(remote_path, local_path, retry=retry - 1)
        else:
            raise FileNotFoundError("下載失?。哼h(yuǎn)程文件 {} 不存在，且重試次數(shù)已用完".format(remote_path))

    def close(self):
        """關(guān)閉連接"""
        if self._connection:
            self._connection.close()
            logging.info("SFTP 連接已關(guān)閉")

if __name__ == '__main__':
    # 請?zhí)鎿Q為真實(shí)的服務(wù)器信息
    host = 'example.com'
    port = 22
    username = 'your_username'
    password = 'your_password'

    download_remote_path = 'outgoing/download.txt'
    download_local_path = 'download.txt'

    upload_local_path = 'upload.txt'
    upload_remote_path = 'incoming/upload.txt'

    # 使用示例
    client = SftpClient(host, port, username, password)

    # 上傳本地文件到遠(yuǎn)程服務(wù)器
    client.upload(upload_local_path, upload_remote_path)

    # 下載遠(yuǎn)程文件到本地（如果文件暫時(shí)不存在會自動(dòng)重試）
    client.download(download_remote_path, download_local_path)

    # 記得關(guān)閉連接
    client.close()