Coverage for /home/runner/work/coxbuild/coxbuild/src/coxbuild/events/__init__.py: 84%

1import asyncio 

2import functools 

3import logging 

4from datetime import date, datetime, time, timedelta 

5from typing import AsyncIterator, Awaitable, Callable, ParamSpec 

6 

7from ..services import EventContext, EventType 

8 

9P = ParamSpec("P") 

10 

11logger = logging.getLogger("events") 

12 

13 

14def onceevent(f: Callable[P, Awaitable[EventContext | None]]) -> Callable[P, EventType]: 

15 """Wrap an awaitable function into event function.""" 

16 @functools.wraps(f) 

17 def wrapper(*args, **kwargs): 

18 async def inner(): 

19 awaitable = f(*args, **kwargs) 

20 result = await awaitable 

21 yield result 

22 return inner() 

23 

24 return wrapper 

25 

26 

27@onceevent 

28async def delay(duration: timedelta): 

29 """Delay occur.""" 

30 await asyncio.sleep(duration.total_seconds()) 

31 

32 

33def now(): 

34 """Occur immediately.""" 

35 return delay(timedelta(seconds=0)) 

36 

37 

38async def repeat(eventBuilder: Callable[[], EventType], repeat: int = 0, continueOnError: bool = False): 

39 """ 

40 Repeat event. 

41 

42 eventBuilder: a callable that return event 

43 repeat: repeat times, 0 for no-repeat, positive integer for finite repeat, negative integer for infinite repeat 

44 continueOnError: continue repeat when error 

45 """ 

46 

47 if repeat >= 0: 

48 repeat += 1 

49 

50 cnt = 0 

51 

52 while repeat != 0: 

53 cnt += 1 

54 logger.debug(f"repeat the {cnt} time.") 

55 try: 

56 async for context in eventBuilder(): 

57 yield context 

58 except Exception as ex: 

59 logger.error(f"Error when repeat the {cnt} time.", exc_info=ex) 

60 if continueOnError: 

61 logger.info(f"Continue repeat on error.") 

62 else: 

63 raise 

64 finally: 

65 if repeat > 0: 

66 repeat -= 1 

67 

68 

69def forever(eventBuilder: Callable[[], EventType]): 

70 """Repeat forever""" 

71 return repeat(eventBuilder, -1) 

72 

73 

74def periodic(period: timedelta): 

75 """Occur forever in an interval.""" 

76 return forever(lambda: delay(period)) 

77 

78 

79async def limit(event: EventType, number: int = 1): 

80 """ 

81 Limit event occur times. 

82 

83 number: limit times, 0 for no-occur, positive integer for finite occur, negative integer for infinite occur 

84 """ 

85 if number == 0: 

86 return 

87 

88 async for context in event: 

89 yield context 

90 

91 if number > 0: 

92 number -= 1 

93 

94 if number == 0: 

95 break 

96 

97 

98def once(event: EventType): 

99 """ 

100 Occur event once. 

101 """ 

102 return limit(event, 1) 

103 

104 

105async def occur(event: EventType): 

106 """ 

107 Create an awaitable for occuring event once. 

108 """ 

109 async for context in event: 

110 return context