How to await a coroutine in pdb

Question

I'm using an async library (asyncpg) and I want to debug some async calls to query the database.

I place a pdb breakpoint and want try out a few queries:

(pdb) await asyncpg.fetch("select * from foo;")
*** SyntaxError: 'await' outside function

It would be great to be able to do this because it would allow me to try out a few SQL queries and see the result, all from the comfort of my debugger.

Is it possible?

Does `asyncio.run(asyncpg.fetch("select * from foo;"))` work? — L3viathan, Nov 15 '19 at 12:04
I doubt it's possible today, given that the asyncio event loop is not reentrant. If your breakpoint is inside an async function, it's possible _in theory_, but highly non-trivial to implement. For the `await` to work, PDB would need to modify execution of a running generator (which is how coroutines are implemented internally) to provide a new yield (await) point. It's comparable to how, given a breakpoint inside a generator, you cannot run `yield bla` from the PDB prompt. — user4815162342, Nov 15 '19 at 12:07
see https://stackoverflow.com/questions/57532678/await-an-async-function-in-python-debugger — amirouche, Jan 14 '20 at 07:09
There is an open feature request for it, but there has been no activity: https://bugs.python.org/issue42045 — shadowtalker, Apr 07 '21 at 15:13

Maarten Derickx · Answer 1 · 2021-06-05T07:34:25.247

I had a similar problem debugging the useage of aiofile. I then found a solution using nest_asyncio. For example if one has the following async example script:

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())

One can then do:

-> print(await afp.read())
(Pdb) loop = asyncio.get_event_loop()
(Pdb) loop.run_until_complete(afp.read())
'Hello world'
(Pdb)

Admittedly it is a bit more tedious then await asyncpg.fetch("select * from foo;") or await afp.read() but it gets the job done. Hopefully a more elegant solution will come up in the future.

score 2 · Answer 2 · answered Jul 25 '22 at 06:16

I want to add on to @M.D.'s answer.

After applying the nest_asyncio to your loop, you can also add the following function:

def return_awaited_value(coroutine: asyncio.coroutine) -> Any:
  
    loop = asyncio.get_event_loop()
    result = loop.run_until_complete(coroutine)
    return result

If you use VSCode as your IDE, then this can be run from VSCode's debug console as

result = return_awaited_value(afp.read())

And VSCode will return it as an object that I find helpful to have while debugging.

score 0 · Answer 3 · answered Jun 14 '21 at 11:50

We can't directly debug the coroutine (since, imagine we have to put the pdb inside the asyncpg.fetch function somewhere by overriding it).

Instead, one possibility is we can make a sync function and convert it to async function using any package like awaits and and 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 )))

How to await a coroutine in pdb

3 Answers3

Linked