# -*- coding: utf-8 -*- """ Abstrakte EventStore-Klasse. Definiert die Schnittstelle für Event-Speicherung. """ from __future__ import annotations from abc import ABC, abstractmethod from dataclasses import dataclass, field from datetime import datetime from typing import TYPE_CHECKING, Any, Iterator import json import uuid if TYPE_CHECKING: from trixy_core.events.event_data.base import EventData @dataclass class StoredEvent: """ Ein gespeichertes Event. """ id: str """Eindeutige ID.""" event_name: str """Name des Events.""" data: dict[str, Any] """Serialisierte Event-Daten.""" timestamp: datetime """Zeitpunkt des Events.""" source: str = "" """Quelle des Events.""" metadata: dict[str, Any] = field(default_factory=dict) """Zusätzliche Metadaten.""" sequence: int = 0 """Sequenznummer für Reihenfolge.""" @classmethod def from_event( cls, event_name: str, event_data: "EventData", sequence: int = 0, ) -> "StoredEvent": """ Erstellt ein StoredEvent aus Event-Daten. Args: event_name: Name des Events. event_data: Die Event-Daten. sequence: Sequenznummer. Returns: StoredEvent. """ # Event-Daten serialisieren data = {} for key, value in event_data.__dict__.items(): if not key.startswith("_"): if isinstance(value, datetime): data[key] = value.isoformat() elif hasattr(value, "__dict__"): data[key] = str(value) else: data[key] = value return cls( id=event_data.event_id, event_name=event_name, data=data, timestamp=event_data.timestamp, source=event_data.source, metadata=dict(event_data.metadata), sequence=sequence, ) def to_dict(self) -> dict[str, Any]: """Konvertiert in ein Dictionary.""" return { "id": self.id, "event_name": self.event_name, "data": self.data, "timestamp": self.timestamp.isoformat(), "source": self.source, "metadata": self.metadata, "sequence": self.sequence, } @classmethod def from_dict(cls, data: dict[str, Any]) -> "StoredEvent": """ Erstellt aus einem Dictionary. Args: data: Das Dictionary. Returns: StoredEvent. """ return cls( id=data["id"], event_name=data["event_name"], data=data["data"], timestamp=datetime.fromisoformat(data["timestamp"]), source=data.get("source", ""), metadata=data.get("metadata", {}), sequence=data.get("sequence", 0), ) def to_json(self) -> str: """Konvertiert in JSON.""" return json.dumps(self.to_dict(), default=str) @classmethod def from_json(cls, json_str: str) -> "StoredEvent": """ Erstellt aus JSON. Args: json_str: Der JSON-String. Returns: StoredEvent. """ return cls.from_dict(json.loads(json_str)) @dataclass class EventQuery: """ Abfrage für gespeicherte Events. """ event_names: list[str] | None = None """Filter nach Event-Namen.""" start_time: datetime | None = None """Start-Zeitpunkt.""" end_time: datetime | None = None """End-Zeitpunkt.""" source: str | None = None """Filter nach Quelle.""" metadata_filter: dict[str, Any] | None = None """Filter nach Metadaten.""" limit: int = 1000 """Maximale Anzahl Ergebnisse.""" offset: int = 0 """Offset für Paginierung.""" order_ascending: bool = True """Sortierreihenfolge.""" after_sequence: int | None = None """Events nach dieser Sequenznummer.""" before_sequence: int | None = None """Events vor dieser Sequenznummer.""" @dataclass class EventQueryResult: """ Ergebnis einer Event-Abfrage. """ events: list[StoredEvent] """Gefundene Events.""" total_count: int """Gesamtzahl (ohne Limit).""" has_more: bool """Ob weitere Events existieren.""" first_sequence: int | None = None """Erste Sequenznummer.""" last_sequence: int | None = None """Letzte Sequenznummer.""" class EventStore(ABC): """ Abstrakte Basisklasse für Event-Speicherung. Speichert Events für Replay, Audit und Analyse. Example: store = InMemoryEventStore() # Event speichern await store.append("user_login", event_data) # Events abfragen query = EventQuery( event_names=["user_login"], start_time=datetime.now() - timedelta(hours=1), ) result = await store.query(query) for event in result.events: print(f"{event.timestamp}: {event.event_name}") """ @abstractmethod async def append( self, event_name: str, event_data: "EventData", ) -> StoredEvent: """ Speichert ein Event. Args: event_name: Name des Events. event_data: Die Event-Daten. Returns: Das gespeicherte Event. """ pass @abstractmethod async def get(self, event_id: str) -> StoredEvent | None: """ Gibt ein Event nach ID zurück. Args: event_id: Die Event-ID. Returns: StoredEvent oder None. """ pass @abstractmethod async def query(self, query: EventQuery) -> EventQueryResult: """ Führt eine Abfrage aus. Args: query: Die Abfrage. Returns: EventQueryResult. """ pass @abstractmethod async def count(self, query: EventQuery | None = None) -> int: """ Zählt Events. Args: query: Optionale Abfrage. Returns: Anzahl der Events. """ pass @abstractmethod async def delete(self, event_id: str) -> bool: """ Löscht ein Event. Args: event_id: Die Event-ID. Returns: True wenn gelöscht. """ pass @abstractmethod async def clear(self) -> int: """ Löscht alle Events. Returns: Anzahl gelöschter Events. """ pass async def get_latest( self, event_name: str | None = None, count: int = 1, ) -> list[StoredEvent]: """ Gibt die neuesten Events zurück. Args: event_name: Optionaler Event-Name-Filter. count: Anzahl der Events. Returns: Liste der Events. """ query = EventQuery( event_names=[event_name] if event_name else None, limit=count, order_ascending=False, ) result = await self.query(query) return result.events async def get_by_sequence_range( self, start: int, end: int, ) -> list[StoredEvent]: """ Gibt Events in einem Sequenzbereich zurück. Args: start: Start-Sequenz (inklusiv). end: End-Sequenz (exklusiv). Returns: Liste der Events. """ query = EventQuery( after_sequence=start - 1, before_sequence=end, limit=end - start, ) result = await self.query(query) return result.events async def stream( self, query: EventQuery | None = None, ) -> Iterator[StoredEvent]: """ Streamt Events. Args: query: Optionale Abfrage. Yields: StoredEvent Objekte. """ query = query or EventQuery() offset = 0 batch_size = 100 while True: batch_query = EventQuery( event_names=query.event_names, start_time=query.start_time, end_time=query.end_time, source=query.source, metadata_filter=query.metadata_filter, limit=batch_size, offset=offset, order_ascending=query.order_ascending, ) result = await self.query(batch_query) for event in result.events: yield event if not result.has_more: break offset += batch_size