service_container.py 12 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382
  1. # -*- coding: utf-8 -*-
  2. """
  3. Service-Container für die Verwaltung aller Trixy-Services.
  4. Verwaltet den Lebenszyklus, Abhängigkeiten und bietet
  5. Zugriff auf registrierte Services.
  6. """
  7. import asyncio
  8. from typing import TYPE_CHECKING, TypeVar, overload
  9. from trixy_core.service.enums import ServiceState
  10. from trixy_core.service.iservice import IService
  11. from trixy_core.utils.debug import pinfo, pdebug, perror, pwarn
  12. if TYPE_CHECKING:
  13. from trixy_core.application import IApplication
  14. T = TypeVar("T", bound=IService)
  15. class ServiceContainer:
  16. """
  17. Container für die Verwaltung aller Services.
  18. Bietet:
  19. - Prioritätsbasierte Startreihenfolge
  20. - Abhängigkeitsauflösung und -injektion
  21. - Gesundheitsprüfungen mit Auto-Restart
  22. - Registry mit __getitem__ Zugriff
  23. """
  24. def __init__(self, application: "IApplication") -> None:
  25. """
  26. Initialisiert den Service-Container.
  27. Args:
  28. application: Referenz zur Hauptanwendung
  29. """
  30. self._application = application
  31. self._services: dict[str, IService] = {}
  32. self._start_order: list[str] = []
  33. self._health_check_interval: float = 30.0
  34. self._health_check_task: asyncio.Task | None = None
  35. self._running = False
  36. @property
  37. def services(self) -> dict[str, IService]:
  38. """Gibt alle registrierten Services zurück."""
  39. return self._services.copy()
  40. def register(self, service_class: type[T]) -> T:
  41. """
  42. Registriert eine Service-Klasse.
  43. Args:
  44. service_class: Die zu registrierende Service-Klasse
  45. Returns:
  46. Die erstellte Service-Instanz
  47. Raises:
  48. ValueError: Wenn der Service bereits registriert ist
  49. """
  50. name = service_class.NAME or service_class.__name__
  51. if name in self._services:
  52. raise ValueError(f"Service '{name}' ist bereits registriert")
  53. service = service_class(self._application)
  54. self._services[name] = service
  55. pdebug(f"Service registriert: {name}")
  56. return service
  57. def register_instance(self, service: IService) -> None:
  58. """
  59. Registriert eine bereits erstellte Service-Instanz.
  60. Args:
  61. service: Die zu registrierende Service-Instanz
  62. Raises:
  63. ValueError: Wenn der Service bereits registriert ist
  64. """
  65. name = service.name
  66. if name in self._services:
  67. raise ValueError(f"Service '{name}' ist bereits registriert")
  68. self._services[name] = service
  69. pdebug(f"Service-Instanz registriert: {name}")
  70. def get_service(self, name: str) -> IService | None:
  71. """
  72. Gibt einen Service nach Namen zurück.
  73. Args:
  74. name: Name des Services
  75. Returns:
  76. Service-Instanz oder None wenn nicht gefunden
  77. """
  78. return self._services.get(name)
  79. def get_service_typed(self, name: str, service_type: type[T]) -> T | None:
  80. """
  81. Gibt einen Service mit Typ-Casting zurück.
  82. Args:
  83. name: Name des Services
  84. service_type: Erwarteter Service-Typ
  85. Returns:
  86. Getypte Service-Instanz oder None
  87. """
  88. service = self._services.get(name)
  89. if service is not None and isinstance(service, service_type):
  90. return service
  91. return None
  92. @overload
  93. def __getitem__(self, key: str) -> IService: ...
  94. @overload
  95. def __getitem__(self, key: type[T]) -> T: ...
  96. def __getitem__(self, key: str | type[T]) -> IService | T:
  97. """
  98. Gibt einen Service per Index-Zugriff zurück.
  99. Args:
  100. key: Service-Name oder Service-Klasse
  101. Returns:
  102. Service-Instanz
  103. Raises:
  104. KeyError: Wenn der Service nicht gefunden wird
  105. """
  106. if isinstance(key, str):
  107. if key not in self._services:
  108. raise KeyError(f"Service '{key}' nicht gefunden")
  109. return self._services[key]
  110. else:
  111. name = key.NAME or key.__name__
  112. if name not in self._services:
  113. raise KeyError(f"Service '{name}' nicht gefunden")
  114. return self._services[name]
  115. def __contains__(self, key: str | type[IService]) -> bool:
  116. """Prüft, ob ein Service registriert ist."""
  117. if isinstance(key, str):
  118. return key in self._services
  119. name = key.NAME or key.__name__
  120. return name in self._services
  121. def _resolve_start_order(self) -> list[str]:
  122. """
  123. Löst die Startreihenfolge basierend auf Priorität und Abhängigkeiten auf.
  124. Returns:
  125. Liste der Service-Namen in Startreihenfolge
  126. Raises:
  127. ValueError: Bei zirkulären Abhängigkeiten oder fehlenden Services
  128. """
  129. # Sortiere nach Priorität
  130. sorted_services = sorted(
  131. self._services.items(),
  132. key=lambda x: x[1].PRIORITY
  133. )
  134. resolved: list[str] = []
  135. unresolved: set[str] = set(self._services.keys())
  136. def resolve(name: str, chain: set[str]) -> None:
  137. """Rekursive Abhängigkeitsauflösung."""
  138. if name in resolved:
  139. return
  140. if name in chain:
  141. raise ValueError(
  142. f"Zirkuläre Abhängigkeit erkannt: {' -> '.join(chain)} -> {name}"
  143. )
  144. if name not in self._services:
  145. raise ValueError(f"Abhängiger Service '{name}' nicht gefunden")
  146. service = self._services[name]
  147. chain = chain | {name}
  148. for dep in service.DEPENDENCIES:
  149. if dep not in self._services:
  150. raise ValueError(
  151. f"Service '{name}' benötigt '{dep}', aber dieser ist nicht registriert"
  152. )
  153. resolve(dep, chain)
  154. resolved.append(name)
  155. unresolved.discard(name)
  156. for name, _ in sorted_services:
  157. if name in unresolved:
  158. resolve(name, set())
  159. return resolved
  160. async def start_all(self) -> None:
  161. """
  162. Startet alle registrierten Services in der richtigen Reihenfolge.
  163. Raises:
  164. Exception: Bei Fehlern während des Starts
  165. """
  166. pinfo("Starte alle Services...")
  167. self._start_order = self._resolve_start_order()
  168. pdebug(f"Startreihenfolge: {self._start_order}")
  169. for name in self._start_order:
  170. await self._start_service(name)
  171. self._running = True
  172. self._health_check_task = asyncio.create_task(self._health_check_loop())
  173. pinfo("Alle Services gestartet")
  174. async def _start_service(self, name: str) -> None:
  175. """
  176. Startet einen einzelnen Service.
  177. Args:
  178. name: Name des zu startenden Services
  179. """
  180. service = self._services[name]
  181. if service.state == ServiceState.RUNNING:
  182. return
  183. pdebug(f"Starte Service: {name}")
  184. service._set_state(ServiceState.STARTING)
  185. try:
  186. await service.on_pre_start()
  187. await service.start()
  188. service._set_state(ServiceState.RUNNING)
  189. await service.on_post_start()
  190. # Benachrichtige abhängige Services
  191. for other_name, other_service in self._services.items():
  192. if name in other_service.DEPENDENCIES:
  193. await other_service.on_dependency_ready(name)
  194. pinfo(f"Service gestartet: {name}")
  195. # Event ausloesen
  196. await self._emit("service_started", {
  197. "service_name": name,
  198. "priority": service.PRIORITY.name if hasattr(service.PRIORITY, "name") else str(service.PRIORITY),
  199. "group": service.GROUP.name if hasattr(service.GROUP, "name") else str(service.GROUP),
  200. })
  201. except Exception as e:
  202. service._set_state(ServiceState.FAILED)
  203. perror(f"Fehler beim Starten von Service '{name}': {e}")
  204. await self._emit("service_error", {
  205. "service_name": name,
  206. "error": str(e),
  207. "traceback": __import__("traceback").format_exc(),
  208. })
  209. raise
  210. async def stop_all(self) -> None:
  211. """Stoppt alle Services in umgekehrter Startreihenfolge."""
  212. pinfo("Stoppe alle Services...")
  213. self._running = False
  214. if self._health_check_task:
  215. self._health_check_task.cancel()
  216. try:
  217. await self._health_check_task
  218. except asyncio.CancelledError:
  219. pass
  220. self._health_check_task = None
  221. # Stoppe in umgekehrter Reihenfolge
  222. for name in reversed(self._start_order):
  223. await self._stop_service(name)
  224. pinfo("Alle Services gestoppt")
  225. async def _emit(self, event_name: str, data: dict) -> None:
  226. """Hilfsmethode zum Ausloesen von Events (falls EventManager verfuegbar)."""
  227. events = getattr(self._application, "events", None)
  228. if events:
  229. try:
  230. await events.emit(event_name, data)
  231. except Exception:
  232. pass
  233. async def _stop_service(self, name: str) -> None:
  234. """
  235. Stoppt einen einzelnen Service.
  236. Args:
  237. name: Name des zu stoppenden Services
  238. """
  239. service = self._services[name]
  240. if service.state == ServiceState.STOPPED:
  241. return
  242. pdebug(f"Stoppe Service: {name}")
  243. service._set_state(ServiceState.STOPPING)
  244. try:
  245. await service.on_pre_stop()
  246. await service.stop()
  247. service._set_state(ServiceState.STOPPED)
  248. await service.on_post_stop()
  249. pinfo(f"Service gestoppt: {name}")
  250. await self._emit("service_stopped", {
  251. "service_name": name,
  252. "reason": "shutdown",
  253. })
  254. except Exception as e:
  255. perror(f"Fehler beim Stoppen von Service '{name}': {e}")
  256. service._set_state(ServiceState.STOPPED)
  257. await self._emit("service_error", {
  258. "service_name": name,
  259. "error": str(e),
  260. "traceback": __import__("traceback").format_exc(),
  261. })
  262. async def restart_service(self, name: str) -> None:
  263. """
  264. Startet einen Service neu.
  265. Args:
  266. name: Name des neu zu startenden Services
  267. Raises:
  268. KeyError: Wenn der Service nicht existiert
  269. """
  270. if name not in self._services:
  271. raise KeyError(f"Service '{name}' nicht gefunden")
  272. pwarn(f"Starte Service neu: {name}")
  273. await self._stop_service(name)
  274. await self._start_service(name)
  275. async def _health_check_loop(self) -> None:
  276. """Führt periodische Gesundheitsprüfungen durch."""
  277. while self._running:
  278. await asyncio.sleep(self._health_check_interval)
  279. if not self._running:
  280. break
  281. for name, service in self._services.items():
  282. if service.state != ServiceState.RUNNING:
  283. continue
  284. try:
  285. healthy = await service.health_check()
  286. if not healthy:
  287. pwarn(f"Service '{name}' Gesundheitsprüfung fehlgeschlagen")
  288. await self.restart_service(name)
  289. except Exception as e:
  290. perror(f"Fehler bei Gesundheitsprüfung von '{name}': {e}")
  291. def set_health_check_interval(self, seconds: float) -> None:
  292. """
  293. Setzt das Intervall für Gesundheitsprüfungen.
  294. Args:
  295. seconds: Intervall in Sekunden
  296. """
  297. self._health_check_interval = seconds