Python+Paramiko编写简单的SFTP文件上传下载工具类
作者：weixin_46244623
在实际项目开发中，我们经常需要通过 SFTP 协议与远程服务器进行文件传输。Paramiko 是 Python 中最流行的 SSH2 协议实现库，基于它我们可以轻松实现 SFTP 客户端功能。
本文将分享一个封装好的 SftpClient 类，支持文件上传、下载（带重试机制）、上传进度提示以及连接管理，代码简洁实用，可直接拷贝到项目中使用。
环境准备
首先确保安装了 paramiko：
pip install paramiko
完整代码实现
from paramiko import Transport, SFTPClient
import time
import errno
import logging
# 配置日志，便于观察上传进度
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):
"""上传进度回调函数"""
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):
"""上传文件，支持进度显示"""
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):
"""判断远程文件是否存在"""
try:
self._connection.stat(remote_path)
return True
except IOError as e:
if e.errno == errno.ENOENT:
return False
raise # 其他错误向上抛出
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("远程文件不存在，{}秒后重试（剩余重试次数：{}）".format(5, retry))
time.sleep(5)
self.download(remote_path, local_path, retry=retry - 1)
else:
raise FileNotFoundError("下载失败：远程文件 {} 不存在，且重试次数已用完".format(remote_path))
def close(self):
"""关闭连接"""
if self._connection:
self._connection.close()
logging.info("SFTP 连接已关闭")
if __name__ == '__main__':
# 请替换为真实的服务器信息
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)
# 上传本地文件到远程服务器
client.upload(upload_local_path, upload_remote_path)
# 下载远程文件到本地（如果文件暂时不存在会自动重试）
client.download(download_remote_path, download_local_path)
# 记得关闭连接
client.close()
功能亮点说明
进度提示：上传时通过 callback 参数实时打印上传进度，便于监控大文件传输。
下载重试机制：在某些场景下（如远程服务器正在生成文件），文件可能暂时不存在。该方法会自动等待并重试 5 次，避免立即失败。
连接管理：使用类变量 _connection 共享连接，避免重复创建 Transport，提高效率。
异常处理：file_exists 方法优雅处理“文件不存在”错误，其他异常正常抛出。
日志友好：使用 logging 输出关键信息，便于生产环境排查问题。
使用注意事项
请务必将密码等敏感信息改为配置文件或环境变量方式读取，不要硬编码在代码中。
如果服务器使用密钥认证，可将 transport.connect 改为 pkey 参数传入私钥。
大文件传输时建议适当调整服务器的超时配置，避免连接中断。
总结
这个 SftpClient 类已经可以满足大部分日常 SFTP 文件传输需求，代码结构清晰、易于扩展。如果需要支持目录递归上传/下载、断点续传等高级功能，可以在此基础上继续封装。
到此这篇关于Python+Paramiko编写简单的SFTP文件上传下载工具类的文章就介绍到这了,更多相关Python SFTP文件上传下载工具类内容请搜索脚本之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持脚本之家！