Python中如何实现SFTP递归列出目录？

阿华AIGC实验室

2026-5-15

这个问题我太有共鸣了——处理上万级别的子目录时，逐个listdir的串行遍历确实会慢到让人抓狂。好消息是，如果你的SFTP服务器是OpenSSH（大部分生产场景都是），它支持一个SFTP扩展，可以通过单次请求递归获取整个目录树，不需要反复发起listdir调用。下面给你两种可行的方案，覆盖Paramiko和其他兼容库：

方案一：用Paramiko调用OpenSSH的readdir扩展

Paramiko底层其实支持OpenSSH的readdir@openssh.com扩展，这个扩展允许我们一次性请求递归列出指定路径下的所有文件和目录。你可以通过直接调用SFTP的底层请求方法来实现，不用自己写递归逻辑：

import paramiko

def recursive_sftp_list(sftp, remote_path):
    # 发送readdir@openssh.com请求，设置递归标志为1（开启递归）
    req = sftp._request(
        "readdir@openssh.com",
        paramiko.SFTPAttributes(),
        remote_path,
        1
    )
    # 接收服务器返回的所有文件/目录属性
    attrs = []
    while True:
        try:
            attr = sftp._read_response(req)
            if attr is None:
                break
            attrs.append(attr)
        except EOFError:
            break
    # 整理结果，过滤掉当前目录(.)和上级目录(..)
    result = []
    for attr in attrs:
        if attr.filename not in ('.', '..'):
            full_path = f"{remote_path.rstrip('/')}/{attr.filename}"
            result.append((full_path, attr))
    return result

# 使用示例
with paramiko.SSHClient() as ssh:
    ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
    ssh.connect('your-host', username='your-user', password='your-pass')
    with ssh.open_sftp() as sftp:
        all_items = recursive_sftp_list(sftp, '/path/to/target-dir')
        # 遍历输出结果，区分文件和目录
        for path, attr in all_items:
            print(f"{path} {'(目录)' if attr.st_mode & 0o40000 else ''}")

这个方法本质上是一次SFTP请求，服务器会把整个递归目录树的所有条目一次性返回，速度会比逐个遍历快几个数量级。需要注意的是，这个扩展仅OpenSSH服务器支持，如果你的SFTP服务器是其他非OpenSSH类型，可能无法兼容。

方案二：用AsyncSSH实现高效递归列表

如果你不局限于Paramiko，AsyncSSH对SFTP扩展的支持更友好，它原生就封装了递归目录列表的能力，代码更简洁易维护：

import asyncssh
import asyncio

async def recursive_list():
    async with asyncssh.connect('your-host', username='your-user', password='your-pass') as conn:
        async with conn.start_sftp_client() as sftp:
            # 直接调用walk，在支持扩展的服务器上会自动使用单次递归请求
            async for path, attr in sftp.walk('/path/to/target-dir', recurse=True):
                print(f"{path} {'(目录)' if attr.is_dir() else ''}")

# 运行异步任务
asyncio.run(recursive_list())

AsyncSSH的walk方法会自动检测服务器是否支持递归扩展，如果支持就用单次请求获取所有数据，效率同样很高；如果不支持，才会降级为客户端侧递归（但这种场景非常少见）。