trixy_core.service

Service Container mit Dependency Injection, Lifecycle-Management und Resilience-Patterns.

Übersicht

from trixy_core.service import (
    # Core
    IService, ServiceContainer,
    ServicePriority, ServiceGroup, ServiceState,
    # Lazy Loading
    LazyService, LazyProxy,
    # Shutdown
    GracefulShutdownManager, ShutdownConfig,
    # Circuit Breaker
    CircuitBreaker, CircuitBreakerConfig, CircuitState,
    # Scopes
    SingletonScope, TransientScope, RequestScope,
    # Metrics
    MetricsCollector, ServiceMetrics,
)

IService

Abstrakte Basisklasse für alle Services.

Definition

from trixy_core.service import IService, ServicePriority, ServiceGroup

class MyService(IService):
    NAME = "my_service"
    PRIORITY = ServicePriority.OPTIONAL
    GROUP = ServiceGroup.UTILITY
    DEPENDENCIES = ["event_manager", "config_manager"]

    async def on_pre_start(self) -> None:
        """Wird vor dem Start aufgerufen."""
        self.logger.debug("Bereite Service vor")

    async def on_start(self) -> None:
        """Hauptstart-Logik."""
        self.logger.info("Service gestartet")

    async def on_post_start(self) -> None:
        """Wird nach erfolgreichem Start aufgerufen."""
        pass

    async def on_pre_stop(self) -> None:
        """Wird vor dem Stoppen aufgerufen."""
        pass

    async def on_stop(self) -> None:
        """Hauptstop-Logik."""
        self.logger.info("Service gestoppt")

    async def on_post_stop(self) -> None:
        """Wird nach dem Stoppen aufgerufen."""
        pass

    async def on_dependency_ready(self, name: str, service: IService) -> None:
        """Wird aufgerufen wenn eine Abhängigkeit bereit ist."""
        self.logger.debug(f"Abhängigkeit {name} bereit")

Klassenattribute

Attribut	Typ	Beschreibung
`NAME`	`str`	Eindeutiger Service-Name
`PRIORITY`	`ServicePriority`	Start-Reihenfolge
`GROUP`	`ServiceGroup`	Kategorisierung
`DEPENDENCIES`	`list[str]`	Benötigte Services

ServicePriority

from trixy_core.service import ServicePriority

class ServicePriority(IntEnum):
    CRITICAL = 0   # System-kritisch (Logger, Config)
    CORE = 10      # Kern-Services (EventManager)
    NETWORK = 20   # Netzwerk-Services (Protocol)
    MANAGER = 30   # Manager (SatelliteManager)
    PLUGIN = 40    # Plugin-Services
    OPTIONAL = 50  # Optionale Services

ServiceGroup

from trixy_core.service import ServiceGroup

class ServiceGroup(Enum):
    CRITICAL = "critical"   # System-kritisch
    NETWORK = "network"     # Netzwerk
    MANAGER = "manager"     # Manager
    AUDIO = "audio"         # Audio-Verarbeitung
    ML = "ml"               # Machine Learning
    UTILITY = "utility"     # Utilities
    PLUGIN = "plugin"       # Plugins

Properties

Property	Typ	Beschreibung
`application`	`Application`	Anwendungsreferenz
`state`	`ServiceState`	Aktueller Zustand
`name`	`str`	Service-Name
`is_running`	`bool`	Service läuft

ServiceContainer

Verwaltet Service-Lifecycle und Abhängigkeiten.

from trixy_core.service import ServiceContainer

container = ServiceContainer(application)

# Service registrieren
container.register(MyService)
container.register(AnotherService)

# Alle Services starten
await container.start_all()

# Service abrufen
my_service = container.get("my_service")

# Service-Status prüfen
if container.is_running("my_service"):
    print("Service läuft")

# Alle Services stoppen
await container.stop_all()

Methoden:

Methode	Beschreibung
`register(service_class)`	Service registrieren
`unregister(name)`	Service entfernen
`get(name)`	Service abrufen
`start_all()`	Alle Services starten
`stop_all()`	Alle Services stoppen
`is_running(name)`	Prüfen ob Service läuft
`get_by_group(group)`	Services nach Gruppe

LazyService

Verzögerte Service-Initialisierung.

from trixy_core.service import LazyService

class ExpensiveService(IService):
    async def on_start(self) -> None:
        # Wird erst bei erstem Zugriff ausgeführt
        self.model = load_large_model()

# Als Lazy registrieren
lazy = LazyService(ExpensiveService)
container.register_lazy("expensive", lazy)

# Service wird erst hier initialisiert
service = await container.get_lazy("expensive")

GracefulShutdownManager

Koordinierter Shutdown mit Timeouts.

from trixy_core.service import (
    GracefulShutdownManager,
    ShutdownConfig,
    ShutdownPhase,
)

config = ShutdownConfig(
    timeout=30.0,           # Gesamttimeout
    phase_timeout=10.0,     # Timeout pro Phase
    force_after=60.0,       # Force-Kill nach
    drain_connections=True,  # Verbindungen abbauen
)

manager = GracefulShutdownManager(container, config)

# Shutdown-Handler registrieren
@manager.on_phase(ShutdownPhase.PRE)
async def pre_shutdown():
    await notify_clients("Server wird heruntergefahren")

@manager.on_phase(ShutdownPhase.CLEANUP)
async def cleanup():
    await flush_caches()

# Shutdown ausführen
report = await manager.shutdown()

print(f"Shutdown abgeschlossen in {report.duration}s")
print(f"Services gestoppt: {report.stopped_count}")
if report.errors:
    for error in report.errors:
        print(f"Fehler: {error}")

ShutdownPhase

Phase	Beschreibung
`PRE`	Vor dem Stoppen
`STOP`	Services stoppen
`CLEANUP`	Aufräumen
`POST`	Nach dem Stoppen

Circuit Breaker

Fault-Tolerance-Pattern für externe Aufrufe.

from trixy_core.service.circuit_breaker import (
    CircuitBreaker,
    CircuitBreakerConfig,
    CircuitState,
    circuit_breaker,
)

# Konfiguration
config = CircuitBreakerConfig(
    failure_threshold=5,    # Fehler bis OPEN
    success_threshold=3,    # Erfolge bis CLOSED
    timeout=30.0,           # Zeit in OPEN
    half_open_max=3,        # Max Versuche in HALF_OPEN
)

# Circuit Breaker erstellen
breaker = CircuitBreaker("api_service", config)

# Verwendung
try:
    async with breaker:
        result = await external_api_call()
except CircuitOpenError:
    # Circuit ist offen - Fallback verwenden
    result = get_cached_result()

Als Decorator

from trixy_core.service.circuit_breaker import circuit_breaker

@circuit_breaker(
    failure_threshold=3,
    timeout=60.0,
    fallback=lambda: {"status": "degraded"}
)
async def call_payment_api(amount: float) -> dict:
    return await payment_client.charge(amount)

CircuitState

CLOSED ──[Fehler >= threshold]──> OPEN
   ^                                 │
   │                                 │
   │                          [timeout]
   │                                 │
   │                                 v
   └──[Erfolge >= threshold]── HALF_OPEN

Zustand	Beschreibung
`CLOSED`	Normal, Requests gehen durch
`OPEN`	Blockiert, Requests werden abgelehnt
`HALF_OPEN`	Testphase, begrenzte Requests

Service Scopes

Verschiedene Lebenszyklen für Services.

from trixy_core.service.scopes import (
    SingletonScope,
    TransientScope,
    RequestScope,
    ScopedScope,
    ScopeContext,
)

# Singleton: Eine Instanz pro Container
singleton = SingletonScope()
container.register(MyService, scope=singleton)

# Transient: Neue Instanz pro Abruf
transient = TransientScope()
container.register(TemporaryService, scope=transient)

# Request: Eine Instanz pro Request-Scope
request_scope = RequestScope()

async def handle_request(request):
    async with ScopeContext("request"):
        # Service-Instanz nur für diesen Request
        service = container.get("scoped_service")
        return await service.process(request)

Scope	Instanzen
`SingletonScope`	Eine pro Container
`TransientScope`	Neue pro Abruf
`RequestScope`	Eine pro Request
`ScopedScope`	Eine pro Scope

Metrics

Service-Metriken für Monitoring.

from trixy_core.service.metrics import (
    MetricsCollector,
    ServiceMetrics,
    Counter, Gauge, Histogram,
    PrometheusReporter,
    JSONReporter,
)

# Collector erstellen
collector = MetricsCollector()

# Metriken definieren
requests = Counter("http_requests_total", "HTTP Requests")
active_connections = Gauge("active_connections", "Aktive Verbindungen")
response_time = Histogram(
    "response_time_seconds",
    "Antwortzeit",
    buckets=[0.01, 0.05, 0.1, 0.5, 1.0, 5.0]
)

# Metriken aufzeichnen
requests.inc()
requests.inc(labels={"method": "GET", "path": "/api"})

active_connections.set(42)
active_connections.inc()
active_connections.dec()

with response_time.time():
    process_request()

# Export
prometheus = PrometheusReporter(collector)
print(prometheus.export())

json_reporter = JSONReporter(collector)
print(json_reporter.export())

ServiceMetrics

Automatische Metriken für Services.

from trixy_core.service.metrics import ServiceMetrics

class MyService(IService):
    async def on_start(self) -> None:
        self.metrics = ServiceMetrics(self.name)

    async def process(self, data):
        with self.metrics.operation("process"):
            # Automatisches Timing und Fehler-Tracking
            return await self._do_process(data)

Dependency Graph

Visualisierung von Service-Abhängigkeiten.

from trixy_core.service import DependencyGraph, GraphFormat

graph = DependencyGraph(container)

# Als Text
print(graph.render(GraphFormat.TEXT))

# Als Mermaid-Diagramm
print(graph.render(GraphFormat.MERMAID))

# Als Graphviz DOT
dot = graph.render(GraphFormat.GRAPHVIZ)
with open("dependencies.dot", "w") as f:
    f.write(dot)

# Als JSON
json_data = graph.render(GraphFormat.JSON)

Beispiel Mermaid-Output:

graph TD
    event_manager --> config_manager
    satellite_manager --> event_manager
    plugin_manager --> event_manager
    plugin_manager --> satellite_manager

service.md 10.0 KB Verlauf Originalformat