'How to use trio_asyncio with legacy sync code in python

I have legacy python application which is synchronous. I started to use async code inside this application in this way (simplified):


async def loader():
  async with trio.open_nursery() as nursery:
    # some async tasks started here
    await trio.to_thread.run_sync(legacyCode)
    

if __name__ == '__main__':
  trio.run(loader)

Inside legacyCode I can use trio.from_thread.run(asyncMethod) to run some async code from the legacy synchronous code. It works well, but now I need to include new library (triopg) which use internally trio_asyncio.

So I need to modify the way how I start my application - I need to replace trio.run by trio_asyncio.run. That's easy but after the trio.to_thread -> trio.from_thread the async code does not work because trio_asyncio has no loop defined.

Here is a short demonstration:

import trio
import trio_asyncio

def main():
  trio.from_thread.run(amain)

async def amain():
  print(f"Loop in amain: {trio_asyncio.current_loop.get()}")  # this print none

async def loader():
  print(f"Loop in loader: {trio_asyncio.current_loop.get()}")  # this print some loop
  await trio.to_thread.run_sync(main)

if __name__ == '__main__':
  trio_asyncio.run(loader)

How should I modify the example above so the trio_asyncio is able to found the loop inside amain() function? Or is this approach completely wrong? If so, how can I use small pieces of async code inside huge synchronous application when libraries needs to use trio and trio_asyncio?

I use python 3.9.



Sources

This article follows the attribution requirements of Stack Overflow and is licensed under CC BY-SA 3.0.

Source: Stack Overflow

Solution Source