我正在使用一个异步库(asyncpg),我想要调试一些异步调用以查询数据库。
我设置了一个pdb断点,想尝试几个查询:
(pdb) await asyncpg.fetch("select * from foo;")
*** SyntaxError: 'await' outside function
如果能够做到这一点的话会很棒,因为它将使我可以尝试一些SQL查询并查看结果,而所有这些都可以在我的调试器舒适的环境中完成。
这种做法可行吗?
我正在使用一个异步库(asyncpg),我想要调试一些异步调用以查询数据库。
我设置了一个pdb断点,想尝试几个查询:
(pdb) await asyncpg.fetch("select * from foo;")
*** SyntaxError: 'await' outside function
如果能够做到这一点的话会很棒,因为它将使我可以尝试一些SQL查询并查看结果,而所有这些都可以在我的调试器舒适的环境中完成。
这种做法可行吗?
我曾经遇到类似的问题,调试aiofile的使用。后来我找到了使用 nest_asyncio 的解决方案。例如,如果有以下异步示例脚本:
import asyncio
from aiofile import async_open
import nest_asyncio
async def main():
async with async_open("/tmp/hello.txt", 'w+') as afp:
await afp.write("Hello ")
await afp.write("world")
afp.seek(0)
breakpoint()
print(await afp.read())
if __name__=="__main__":
loop = asyncio.get_event_loop()
nest_asyncio.apply(loop)
loop.run_until_complete(main())
接下来可以执行以下操作:
-> print(await afp.read())
(Pdb) loop = asyncio.get_event_loop()
(Pdb) loop.run_until_complete(afp.read())
'Hello world'
(Pdb)
诚然,这比await asyncpg.fetch("select * from foo;")
或者await afp.read()
要繁琐一些,但它能完成工作。希望未来会出现更加优雅的解决方案。
def return_awaited_value(coroutine: asyncio.coroutine) -> Any:
loop = asyncio.get_event_loop()
result = loop.run_until_complete(coroutine)
return result
result = return_awaited_value(afp.read())
VSCode将返回此对象,并且我发现在调试时非常有用。我们无法直接调试协程(因为,想象一下,我们必须通过覆盖asyncpg.fetch
函数中的某个位置来放置pdb)。
相反,一个可能性是我们可以创建一个同步函数,并使用任何类似awaits和pdb的包将其转换为异步函数:
import asyncio from awaits.awaitable import awaitable
@awaitable
def overrided_fectch_function( a , b ):
print(a)
import pdb; pdb.set_trace()
return a + b
print(b)
# Now sum is a coroutine! While it is running in a separate thread, control is passed to the event-loop.
print ( asyncio . run ( overrided_fectch_function ( 2 , 2 )))
asyncio.run(asyncpg.fetch("select * from foo;"))
能正常工作吗? - L3viathanawait
正常工作,PDB需要修改正在运行的生成器(这是协程的内部实现方式)的执行方式,以提供新的yield(await)点。这类似于在生成器内设置断点时无法从PDB提示符中运行“yield bla”的情况。 - user4815162342