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?

Predesignate

Does asyncio.run(asyncpg.fetch("select * from foo;")) work? – Illustrative

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. – Shrill

see #57533178 – Indogermanic

There is an open feature request for it, but there has been no activity: bugs.python.org/issue42045 – Inviolate

Answer 1

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.

Canadianism

Answer 2

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:

import asyncio
import typing

def return_awaited_value(coroutine: asyncio.coroutines) -> typing.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.

Defamation

Answer 3

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

Radioscope