Coverage for /home/runner/work/coxbuild/coxbuild/src/coxbuild/runners.py: 100%

1import asyncio 

2import inspect 

3from datetime import timedelta 

4from timeit import default_timer as timer 

5from typing import Awaitable, Callable 

6 

7 

8class Runner: 

9 """Generic async runner.""" 

10 

11 def __init__(self, func: Callable[[], Awaitable | None]) -> None: 

12 self.func = func 

13 self.duration = timedelta() 

14 self.exc_type = None 

15 self.exc_value = None 

16 self.exc_tb = None 

17 

18 async def __aenter__(self) -> Callable[[], Awaitable | None]: 

19 self._tic = timer() 

20 

21 return self.func 

22 

23 async def __aexit__(self, exc_type, exc_value, exc_tb) -> bool: 

24 self.exc_type = exc_type 

25 self.exc_value = exc_value 

26 self.exc_tb = exc_tb 

27 

28 self.duration = timedelta(seconds=timer()-self._tic) 

29 

30 del self._tic 

31 return True 

32 

33 def __await__(self): 

34 async def wrapper(): 

35 async with self as run: 

36 res = run() 

37 if inspect.isawaitable(res): 

38 res = await res 

39 yield from wrapper().__await__()