Source code for ewoksjob.events.readers.sqlite3

import sqlite3
from typing import Iterator
from ewoksutils.event_utils import FIELD_TYPES
from ewoksutils import sqlite3_utils
from .base import EwoksEventReader, EventType


[docs] class Sqlite3EwoksEventReader(EwoksEventReader): def __init__(self, uri: str, **_) -> None: super().__init__() self._uri = uri self.__connection = None self.__sql_types = sqlite3_utils.python_to_sql_types(FIELD_TYPES)
[docs] def close(self): if self.__connection is not None: self.__connection.close() self.__connection = None super().close()
@property def _connection(self): if self.__connection is None: self.__connection = sqlite3.connect(self._uri, uri=True) return self.__connection
[docs] def wait_events(self, **kwargs) -> Iterator[EventType]: yield from self.poll_events(**kwargs)
[docs] def get_events(self, **filters) -> Iterator[EventType]: yield from sqlite3_utils.select( self._connection, "ewoks_events", field_types=FIELD_TYPES, sql_types=self.__sql_types, **filters )