在 pycharm 调试器中调试异步程序

Mic*_*ein 5 python debugging asynchronous python-multithreading python-asyncio

我的请求是从调试器和 pycharm 提示符中获取调试数据:

假设我有一个带有错误的旧式 python 同步程序:

   def listdir(self, remote_path):
         with ssh.connect(self.connection_settings['host'], username=self.connection_settings['username'],
                                    password=self.connection_settings['password']) as conn:
            with conn.start_sftp_client() as sftp:
                sftp.chdir(remote_path) # breakpoint here

                files = sftp.readdir()
                files_name = [file.filename for file in files]

                return files_name

Run Code Online (Sandbox Code Playgroud)

我会停止它,转到调试器,单击调试器控制台-> 显示 python 提示,然后在运行时调试我的问题:

> sftp.getcwd()
bad_user_location
> sftp.chdir(good_base_location)
> sftp.readdir()
Run Code Online (Sandbox Code Playgroud)

只需即时重写代码并测试状态和运行时即可。(这一直是我调试的方式,使用immediate windowMicrosoft 语言开发中的,以及 Python 和其他脚本语言中的 REPL。

现在我有一个异步程序:

  async def listdir(self, remote_path):
        async with asyncssh.connect(self.connection_settings['host'], username=self.connection_settings['username'],
                                    password=self.connection_settings['password']) as conn:
            async with conn.start_sftp_client() as sftp:
                await sftp.chdir(remote_path)  # breakpoint here


                files = await sftp.readdir()
                files_name = [file.filename for file in files]

                return files_name

Run Code Online (Sandbox Code Playgroud)

我停止它,转到调试器,单击调试器控制台-> 显示 python 提示,然后在运行时调试我的问题:

> sftp.getcwd()
<CoroWrapper SFTPClient.getcwd() running at....truncated info
Run Code Online (Sandbox Code Playgroud)
  1. 我尝试获取事件循环并在其上运行,但我无法在没有函数的情况下等待。
  2. 我安装了允许 event_loop 的 IPython 7,但它不会运行任务,只是添加到事件循环中,因为它是为 jupyter 而不是这个用例构建的
  3. 我尝试创建一个新的事件循环,但一个线程中只能有一个。
  4. 我尝试使用 nest_asyncio 并应用,但是 >nest_asyncio.events.new_event_loop().create_task(sftp.getcwd()) 但它只是将它添加到循环中,它没有运行它。<Task pending coro=<SFTPClient.getcwd() running at...
  5. 我认为我最好的办法是创建第二个线程,在那里创建一个新循环,然后等待该循环中的任务。类似于关于 2 个循环的问题,但我对此有困难。就像是:
   >d.run_sync(sftp.getcwd())
   directory_name
   >d.run_sync(sft.chdir(somewhere_else))
   0 # exit code
Run Code Online (Sandbox Code Playgroud)

我在频道 v2.0 中在Andrew Gadwin 的精彩文章下发现了类似的内容

   >asgiref.sync.async_to_sync(sftp.getcwd())
   ><asgiref.sync.AsyncToSync object at 0x7ff94c7b37e0> ... created at <input>:1> was never yielded from
   >ERROR:asyncio:<CoroWrapper SFTPClient.getcwd()
Run Code Online (Sandbox Code Playgroud)
  1. 也许我可以将 jupyternotebook 附加到我正在运行的调试器,但这将是我最后的手段,因为它并不是一个很好的可维护解决方案

我认为能够在运行时调试协同程序至关重要,因此我非常重视您的解决方案。

谢谢

Ray*_*nda 1

对调试器的异步支持 [实验功能]

调试控制台允许您输入异步调用并在函数外部使用await关键字来在调试时检查异步代码。

在此输入图像描述