Coverage for async_durable_execution/_core/task.py: 100%

1"""Async task helpers used by durable operations.""" 

2 

3from __future__ import annotations 

4 

5import asyncio 

6from collections.abc import Callable, Coroutine 

7from typing import Any, TypeVar 

8 

9 

10T = TypeVar("T") 

11 

12 

13def create_eager_task( 

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

15) -> asyncio.Task[T]: 

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

17 loop = asyncio.get_running_loop() 

18 coro = coro_factory() 

19 

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

21 if eager_task_factory is not None: 

22 return eager_task_factory(loop, coro) 

23 

24 return loop.create_task(coro)