Mike O'Connor
08/02/2023, 9:18 AMimport asyncio
import pytest
from prefect import flow, task
@task
async def task_a():
return id(asyncio.get_event_loop())
@task
async def task_b():
return id(asyncio.get_event_loop())
@flow
async def some_flow():
loop_a = await task_a()
loop_b = await task_b()
return loop_a, loop_b
@pytest.mark.asyncio
async def test_task_event_loops():
loop_a, loop_b = await some_flow()
print(f"{loop_a=}")
print(f"{loop_b=}")
assert loop_a == loop_b
Konrad Maliszewski
08/02/2023, 9:29 AMThomas Weatherston
08/02/2023, 11:40 AMJake Kaplan
08/02/2023, 12:58 PMMike O'Connor
08/21/2023, 1:22 PMConor
09/02/2023, 8:20 PM