从Paramiko同步上传回调调用异步函数失败求助

问题原因 #

你碰到的RuntimeWarning: coroutine 'wait' was never awaited错误，核心问题是：同步的upload_callback函数里调用了协程函数asyncio.wait，但同步环境无法用await触发它执行。而且asyncio.wait在这里完全多余——loop.create_task已经把异步任务提交给事件循环了，不需要额外调用wait。

修复后的代码 #

直接移除asyncio.wait调用，仅保留loop.create_task提交异步任务即可：

import os
import paramiko
import asyncio

async def async_websocket_call(msg):
    print(msg)

async def test_paramiko():
    ssh = paramiko.SSHClient()
    ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
    ssh.connect("web.demo.net", username="root", key_filename=os.environ['USERPROFILE'] + "\\.ssh\\id_rsa")

    def upload_callback(sofar, total):
        loop = asyncio.get_running_loop()
        # 直接把异步任务提交给事件循环，无需wait
        loop.create_task(async_websocket_call(f"Uploaded: {100*(sofar/total):.1f}%"))
        
    sftp = ssh.open_sftp()
    sftp.put('test.tar.gz', '/var/www/html/test.tar.gz', upload_callback)
    sftp.close()

    ssh.close()

loop = asyncio.get_event_loop()
loop.run_until_complete(test_paramiko())
loop.close()

额外优化说明 #

1. 避免阻塞事件循环：sftp.put是同步阻塞调用，会卡住整个asyncio事件循环直到上传完成。如果需要不阻塞事件循环，可以把上传操作放到线程池执行：

async def test_paramiko():
    ssh = paramiko.SSHClient()
    ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
    ssh.connect("web.demo.net", username="root", key_filename=os.environ['USERPROFILE'] + "\\.ssh\\id_rsa")

    def upload_with_callback():
        def upload_callback(sofar, total):
            loop = asyncio.get_running_loop()
            loop.create_task(async_websocket_call(f"Uploaded: {100*(sofar/total):.1f}%"))
        
        sftp = ssh.open_sftp()
        sftp.put('test.tar.gz', '/var/www/html/test.tar.gz', upload_callback)
        sftp.close()

    loop = asyncio.get_running_loop()
    # 将同步上传操作放入线程池，避免阻塞事件循环
    await loop.run_in_executor(None, upload_with_callback)

    ssh.close()

2. 关于asyncssh的适配：你提到asyncssh的SFTPProgressHandler是同步回调，同样可以用上述方法解决——在同步回调内部通过asyncio.get_running_loop().create_task提交异步任务即可，不需要修改回调的类型定义。

内容的提问来源于stack exchange，提问作者Ray Hulha