Backup-bdg/OpenHands

1from abc import abstractmethod2from itertools import islice3from typing import Iterable4 5from deprecated import deprecated  # type: ignore6 7from openhands.events.event import Event, EventSource8from openhands.events.event_filter import EventFilter9 10 11class EventStoreABC:12    """13    A stored list of events backing a conversation14    """15 16    sid: str17    user_id: str | None18 19    @abstractmethod20    def search_events(21        self,22        start_id: int = 0,23        end_id: int | None = None,24        reverse: bool = False,25        filter: EventFilter | None = None,26        limit: int | None = None,27    ) -> Iterable[Event]:28        """29        Retrieve events from the event stream, optionally excluding events using a filter30 31        Args:32            start_id: The ID of the first event to retrieve. Defaults to 0.33            end_id: The ID of the last event to retrieve. Defaults to the last event in the stream.34            reverse: Whether to retrieve events in reverse order. Defaults to False.35            filter: An optional event filter36 37        Yields:38            Events from the stream that match the criteria.39        """40 41    @deprecated('Use search_events instead')42    def get_events(43        self,44        start_id: int = 0,45        end_id: int | None = None,46        reverse: bool = False,47        filter_out_type: tuple[type[Event], ...] | None = None,48        filter_hidden: bool = False,49    ) -> Iterable[Event]:50        yield from self.search_events(51            start_id,52            end_id,53            reverse,54            EventFilter(exclude_types=filter_out_type, exclude_hidden=filter_hidden),55        )56 57    @abstractmethod58    def get_event(self, id: int) -> Event:59        """Retrieve a single event from the event stream. Raise a FileNotFoundError if there was no such event"""60 61    @abstractmethod62    def get_latest_event(self) -> Event:63        """Get the latest event from the event stream"""64 65    @abstractmethod66    def get_latest_event_id(self) -> int:67        """Get the id of the latest event from the event stream"""68 69    @deprecated('use search_events instead')70    def filtered_events_by_source(self, source: EventSource) -> Iterable[Event]:71        yield from self.search_events(filter=EventFilter(source=source))72 73    @deprecated('use search_events instead')74    def get_matching_events(75        self,76        query: str | None = None,77        event_types: tuple[type[Event], ...] | None = None,78        source: str | None = None,79        start_date: str | None = None,80        end_date: str | None = None,81        start_id: int = 0,82        limit: int = 100,83        reverse: bool = False,84    ) -> list[Event]:85        """Get matching events from the event stream based on filters.86 87        Args:88            query: Text to search for in event content89            event_types: Filter by event type classes (e.g., (FileReadAction, ) ).90            source: Filter by event source91            start_date: Filter events after this date (ISO format)92            end_date: Filter events before this date (ISO format)93            start_id: Starting ID in the event stream. Defaults to 094            limit: Maximum number of events to return. Must be between 1 and 100. Defaults to 10095            reverse: Whether to retrieve events in reverse order. Defaults to False.96 97        Returns:98            list: List of matching events (as dicts)99        """100        if limit < 1 or limit > 100:101            raise ValueError('Limit must be between 1 and 100')102 103        events = self.search_events(104            start_id=start_id,105            reverse=reverse,106            filter=EventFilter(107                query=query,108                include_types=event_types,109                source=source,110                start_date=start_date,111                end_date=end_date,112            ),113        )114        return list(islice(events, limit))115