I am creating an async python function during runtime. It can be created as a plain string / python ast object. Then I execute the code using exec().
I want to run the generated code and wait till the execution ends for it. I generate the function, and then inejcting code so that the final code looks like this-
import asyncio
async def my_app():
# some async calls etc
loop = asyncio.get_event_loop()
task = loop.create_task(my_app())
In my caller code, I am doing
exec(my_code)
await asyncio.sleep(0)
I was having trouble doing await my_app() in the python code (caller is already an async function).
According to my understanding, the sleep(0) gives control to the event loop to process other tasks. But I am not sure how this will work out if the code generated makes multiple async calls (which it possibly will). How to assure the exectuon finished.
If you really do need to: generate an async function, exec it in a namespace, grab the reference to the function, and run it as usual.
import asyncio
generated_code = """
import asyncio
async def my_app():
print("hello!")
await asyncio.sleep(1)
print("hoi!")
# some async calls etc
"""
async def go():
ns = {}
exec(generated_code, ns)
my_task = asyncio.create_task(ns["my_app"]())
await my_task
asyncio.run(go())
EDIT: Re the comments; instead of global variables, pass in an object to modify (a dataclass here for brevity's sake):
import asyncio
from dataclasses import dataclass
generated_code = """
import asyncio
async def my_app(state):
print("hello!")
state.age *= 2
state.name = state.name[::-1]
await asyncio.sleep(1)
print("hoi!")
"""
@dataclass
class MyState:
name: str
age: int
async def go():
state = MyState(name="Arthur", age=42)
ns = {}
exec(generated_code, ns)
print("State before", state)
my_task = asyncio.create_task(ns["my_app"](state))
await my_task
print("State after", state)
asyncio.run(go())
This outputs
State before MyState(name='Arthur', age=42)
hello!
hoi!
State after MyState(name='ruhtrA', age=84)