Coverage for async-durable-execution/src/async_durable_execution/task.py: 100%

1from __future__ import annotations 

2 

3import asyncio 

4from collections.abc import Callable, Coroutine 

5from typing import Any, TypeVar 

6 

7 

8T = TypeVar("T") 

9 

10 

11def create_eager_task( 

12 coro_factory: Callable[[], Coroutine[Any, Any, T]], 

13) -> asyncio.Task[T]: 

14 """Create an operation task, using eager start when the runtime supports it.""" 

15 loop = asyncio.get_running_loop() 

16 coro = coro_factory() 

17 

18 eager_task_factory = getattr(asyncio, "eager_task_factory", None) 

19 if eager_task_factory is not None: 

20 return eager_task_factory(loop, coro) 

21 

22 return loop.create_task(coro)