diff --git a/olymp/events.py b/olymp/events.py index 053fd6f..aa7973e 100644 --- a/olymp/events.py +++ b/olymp/events.py @@ -23,27 +23,6 @@ _LOGGER = logging.getLogger(__name__) _SUBSCRIBERS: list[EventSubscriber] = [] _SUBSCRIBERS_LOCK = threading.Lock() -_LIST_EVENTS_QUERY = """ - SELECT * - FROM events - WHERE (? IS NULL OR event_type = ?) - AND (? IS NULL OR run_id = ?) - AND (? IS NULL OR plan_id = ?) - AND (? IS NULL OR node_id = ?) - ORDER BY event_id ASC - LIMIT ? -""" -_LIST_EVENTS_AFTER_QUERY = """ - SELECT * - FROM events - WHERE event_id > ? - AND (? IS NULL OR event_type = ?) - AND (? IS NULL OR run_id = ?) - AND (? IS NULL OR plan_id = ?) - AND (? IS NULL OR node_id = ?) - ORDER BY event_id ASC - LIMIT ? -""" def subscribe_events(handler: EventSubscriber) -> Callable[[], None]: @@ -149,23 +128,14 @@ def list( checked_run = _nullable_text(run_id, "run_id") checked_plan = _nullable_text(plan_id, "plan_id") checked_node = _nullable_text(node_id, "node_id") - filter_values: tuple[object, ...] = ( - checked_type, - checked_type, - checked_run, - checked_run, - checked_plan, - checked_plan, - checked_node, - checked_node, - checked_limit, + query, values = _event_list_query( + after_event_id=checked_after, + limit=checked_limit, + event_type=checked_type, + run_id=checked_run, + plan_id=checked_plan, + node_id=checked_node, ) - if checked_after is None: - query = _LIST_EVENTS_QUERY - values = filter_values - else: - query = _LIST_EVENTS_AFTER_QUERY - values = (checked_after, *filter_values) with closing(self._connect()) as db: rows = db.execute(query, values).fetchall() return [_event_record(row) for row in rows] @@ -185,6 +155,36 @@ def emit_event(path: Path, event_type: str, **kwargs: Any) -> dict[str, Any]: return EventStore(path).append(event_type, **kwargs) +def _event_list_query( + *, + after_event_id: int | None, + limit: int, + event_type: str | None, + run_id: str | None, + plan_id: str | None, + node_id: str | None, +) -> tuple[str, tuple[object, ...]]: + clauses: list[str] = [] + values: list[object] = [] + if after_event_id is not None: + clauses.append("event_id > ?") + values.append(after_event_id) + for clause, value in ( + ("event_type = ?", event_type), + ("run_id = ?", run_id), + ("plan_id = ?", plan_id), + ("node_id = ?", node_id), + ): + if value is not None: + clauses.append(clause) + values.append(value) + where = f"WHERE {' AND '.join(clauses)}" if clauses else "" + values.append(limit) + # Clause text comes only from the fixed literals above; every external value is bound. + query = "\n".join(("SELECT *", "FROM events", where, "ORDER BY event_id ASC", "LIMIT ?")) + return query, tuple(values) + + def _notify_subscribers(event: dict[str, Any]) -> None: with _SUBSCRIBERS_LOCK: subscribers = tuple(_SUBSCRIBERS) diff --git a/tests/test_events.py b/tests/test_events.py index f4742e4..ac7427c 100644 --- a/tests/test_events.py +++ b/tests/test_events.py @@ -88,18 +88,42 @@ def test_event_store_filters_and_paginates(self) -> None: limit=1, ) injected_filter = store.list(event_type="plugin.test' OR 1=1 --") + cursor_query, cursor_values = events_module._event_list_query( + after_event_id=1, + limit=100, + event_type=None, + run_id=None, + plan_id=None, + node_id=None, + ) + run_query, run_values = events_module._event_list_query( + after_event_id=None, + limit=100, + event_type=None, + run_id="run-a", + plan_id=None, + node_id=None, + ) with sqlite3.connect(db_path) as db: - query_plan = db.execute( - "EXPLAIN QUERY PLAN " + events_module._LIST_EVENTS_AFTER_QUERY, - (1, None, None, None, None, None, None, None, None, 100), + cursor_plan = db.execute( + "EXPLAIN QUERY PLAN " + cursor_query, + cursor_values, + ).fetchall() + run_plan = db.execute( + "EXPLAIN QUERY PLAN " + run_query, + run_values, ).fetchall() self.assertEqual([event["event_id"] for event in first_page], matching_ids[:1]) self.assertEqual([event["event_id"] for event in second_page], matching_ids[1:]) self.assertEqual(injected_filter, []) self.assertTrue( - any("SEARCH events USING INTEGER PRIMARY KEY" in str(row[3]) for row in query_plan), - query_plan, + any("SEARCH events USING INTEGER PRIMARY KEY" in str(row[3]) for row in cursor_plan), + cursor_plan, + ) + self.assertTrue( + any("SEARCH events USING INDEX idx_events_run_id" in str(row[3]) for row in run_plan), + run_plan, ) def test_subscriber_failure_is_isolated_and_secret_safe(self) -> None: