Coverage for /home/runner/work/coxbuild/coxbuild/src/coxbuild/services.py: 75%

1import asyncio 

2import inspect 

3import logging 

4import sys 

5import traceback 

6from dataclasses import dataclass, field 

7from typing import TYPE_CHECKING, Any, AsyncIterator, Awaitable, Callable 

8 

9from coxbuild.configurations import Configuration 

10from coxbuild.exceptions import (CoxbuildRuntimeException, 

11 CoxbuildSchemaException) 

12from coxbuild.runners import Runner 

13from coxbuild.runtime import ExecutionState 

14from coxbuild.tasks import Task, named, task 

15 

16if TYPE_CHECKING: 

17 from .extensions import Extension 

18 

19logger = logging.getLogger("services") 

20 

21 

22class EventFailed(CoxbuildRuntimeException): 

23 """Event failed.""" 

24 pass 

25 

26 

27class EventHandlerFailed(CoxbuildRuntimeException): 

28 """Event handler failed.""" 

29 pass 

30 

31 

32@dataclass 

33class EventContext: 

34 """Event occur information.""" 

35 args: list[Any] = field(default_factory=list) 

36 """arguments""" 

37 kwds: dict[str, Any] = field(default_factory=dict) 

38 """keyword arguments""" 

39 

40 @classmethod 

41 def build(cls, *args: Any, **kwds: Any) -> "EventContext": 

42 return EventContext(args, kwds) 

43 

44 

45EventType = AsyncIterator[EventContext | None] 

46 

47 

48@dataclass 

49class EventHandler: 

50 """Handler for a event.""" 

51 event: EventType 

52 """event generator, when the event occurs, the awaitable return""" 

53 handler: Task 

54 """event handler""" 

55 safe: bool = False 

56 """prevent exception""" 

57 name: str = "" 

58 """handler name""" 

59 extension: "Extension | None" = None 

60 

61 async def handle(self, config: Configuration | None = None): 

62 logger.debug(f"Handle for event: {self.name}.") 

63 

64 config = config or Configuration() 

65 executionState = ExecutionState(config) 

66 

67 try: 

68 async for context in self.event: 

69 oldhandler = executionState.handler 

70 oldevent = executionState.event 

71 

72 executionState.handler = self 

73 executionState.event = context 

74 

75 logger.debug(f"Event occurs: {self.name}({context}).") 

76 

77 logger.debug(f"Event handling: {self.name}({context}).") 

78 

79 try: 

80 runner = self.handler( 

81 *context.args, **context.kwds) if context else self.handler() 

82 runner.context.config = config 

83 result = await runner 

84 except Exception as ex: 

85 if self.safe: 

86 print( 

87 f"Exception when event handler handling {self.name}({context}).") 

88 traceback.print_exception(ex, file=sys.stdout) 

89 else: 

90 raise EventHandlerFailed( 

91 f"Exception when event handler handling {self.name}({context})", ex) 

92 finally: 

93 executionState.handler = oldhandler 

94 executionState.event = oldevent 

95 

96 logger.debug(f"Event handled: {self.name}({context}).") 

97 except Exception as ex: 

98 logger.error( 

99 f"Event handler '{self.name}' failed.", exc_info=ex) 

100 

101 if self.safe: 

102 print(f"Exception in event handler {self.name}.") 

103 traceback.print_exception(ex, file=sys.stdout) 

104 else: 

105 raise EventFailed( 

106 f"Exception in event handler {self.name}", ex) 

107 

108 logger.debug(f"Finish handle for event: {self.name}.") 

109 

110 

111@dataclass 

112class Service: 

113 """Event-based service.""" 

114 

115 handlers: list[EventHandler] = field(default_factory=list) 

116 """handlers in the service""" 

117 

118 def copy(self) -> "Service": 

119 """Copy the service.""" 

120 return Service(self.handlers.copy()) 

121 

122 def register(self, handler: EventHandler): 

123 """Register handler.""" 

124 self.handlers.append(handler) 

125 logger.debug(f"Register event handler: {handler}") 

126 

127 def __call__(self, *args: Any, **kwds: Any) -> Any: 

128 return ServiceRunner(self) 

129 

130 

131@dataclass 

132class ServiceContext: 

133 """Service context.""" 

134 service: Service 

135 """service""" 

136 config: Configuration | None = None 

137 """configuration""" 

138 

139 

140class ServiceRunner(Runner): 

141 def __init__(self, service: Service) -> None: 

142 self.service = service 

143 self.context = ServiceContext(self.service) 

144 super().__init__(self._run) 

145 

146 async def _run(self): 

147 logger.debug("Run service.") 

148 await asyncio.gather(*[e.handle(self.context.config) for e in self.service.handlers]) 

149 logger.debug("Finish service.") 

150 

151 

152def on(event: Callable[[], Awaitable], safe: bool = False, name: str | None = None): 

153 """ 

154 Decorator to register event handler. 

155 

156 event: event generator, when the event occurs, the awaitable return 

157 handler: event handler 

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

159 safe: prevent exception 

160 name: handler name, None to use function name 

161 """ 

162 def decorator(handler: Callable[[], None] | Task | EventHandler | list[EventHandler]) -> EventHandler | list[EventHandler]: 

163 if isinstance(handler, EventHandler): 

164 tname = name or handler.name 

165 tk = handler.handler 

166 elif isinstance(handler, list): 

167 if len(handler) == 0: 

168 raise CoxbuildSchemaException("No event handler in list.") 

169 tname = name or handler[0].name 

170 tk = handler[0].handler 

171 elif isinstance(handler, Task): 

172 tname = name or handler.name 

173 tk = handler 

174 else: 

175 tname = name or handler.__name__ 

176 tk = named(tname)(task(handler)) 

177 

178 eh = EventHandler(event, tk, safe, tname) 

179 

180 if isinstance(handler, EventHandler): 

181 return [eh, handler] 

182 elif isinstance(handler, list): 

183 handler.append(eh) 

184 return handler 

185 else: 

186 return eh 

187 

188 return decorator