perf(store): SQLite loop-safety, connection lifecycle & query pushdown

custom_components/hass_datapoints/__init__.py

@@ -258,7 +258,8 @@ async def async_unload_entry(hass: HomeAssistant, entry: DatapointsConfigEntry)
     """Unload a config entry."""
     unload_ok = await hass.config_entries.async_unload_platforms(entry, PLATFORMS)
     hass.services.async_remove(DOMAIN, SERVICE_RECORD)
-    hass.data[DOMAIN].pop("store", None)
+    if store := hass.data[DOMAIN].pop("store", None):
+        await hass.async_add_executor_job(store.close)
     hass.data[DOMAIN].pop("anomaly_cache", None)
     hass.data[DOMAIN].pop("in_flight", None)
     hass.data[DOMAIN].pop("scan_semaphore", None)

custom_components/hass_datapoints/hass-datapoints-cards.js

@@ -42252,7 +42252,7 @@
 	].forEach((card) => {
 		if (!registeredTypes.has(card.type)) window.customCards?.push(card);
 	});
-	console.groupCollapsed(`%c hass-datapoints %c v0.6.3 loaded `, "color:#fff;background:#03a9f4;font-weight:bold;padding:2px 6px;border-radius:3px 0 0 3px", "color:#03a9f4;background:#fff;font-weight:bold;padding:2px 6px;border:1px solid #03a9f4;border-radius:0 3px 3px 0", ...[]);
+	console.groupCollapsed(`%c hass-datapoints %c v0.6.4 loaded `, "color:#fff;background:#03a9f4;font-weight:bold;padding:2px 6px;border-radius:3px 0 0 3px", "color:#03a9f4;background:#fff;font-weight:bold;padding:2px 6px;border:1px solid #03a9f4;border-radius:0 3px 3px 0", ...[]);
 	console.log("Enable debug logging by setting %cwindow.__HASS_DATAPOINTS_DEV__ = true", "color:#333;background:#eee;border:1px solid #777;padding:2px 6px;border-radius:5px; font-family: Courier");
 	console.groupEnd();
 	//#endregion

custom_components/hass_datapoints/manifest.json

@@ -8,5 +8,5 @@
   "iot_class": "local_push",
   "issue_tracker": "https://github.com/buggedcom/hass-datapoints/issues",
   "requirements": [],
-  "version": "0.6.3"
+  "version": "0.6.4"
 }

custom_components/hass_datapoints/sensor.py

@@ -95,6 +95,11 @@ class _DatapointsSensorBase(SensorEntity):
     """Shared base for all Hass Data Points sensor entities."""
 
     _attr_has_entity_name = True
+    # Subclasses whose _compute() performs blocking I/O (SQLite reads) set this
+    # to True so refreshes run in the executor instead of on the event loop.
+    # Sensors that compute from in-memory monitor state leave it False and
+    # refresh inline (offloading those would race with loop-side mutations).
+    _compute_blocks_io = False
 
     def __init__(self, entry: ConfigEntry, store: DatapointsStore) -> None:
         """Initialise with config entry and data store."""
@@ -118,7 +123,19 @@ async def async_added_to_hass(self) -> None:
 
     def _handle_store_update(self) -> None:
         """Refresh state after any store mutation."""
-        self._attr_native_value = self._compute()
+        self._refresh_value()
+
+    def _refresh_value(self) -> None:
+        """Recompute and write state — off the loop when _compute() blocks."""
+        if self._compute_blocks_io:
+            self.hass.async_create_task(self._async_refresh_value())
+        else:
+            self._attr_native_value = self._compute()
+            self.async_write_ha_state()
+
+    async def _async_refresh_value(self) -> None:
+        """Run the blocking _compute() in the executor, then write state."""
+        self._attr_native_value = await self.hass.async_add_executor_job(self._compute)
         self.async_write_ha_state()
 
     def _compute(self) -> Any:
@@ -148,8 +165,7 @@ async def async_added_to_hass(self) -> None:
     @callback
     def _handle_time_interval(self, now: datetime) -> None:
         """Refresh state on every timer tick."""
-        self._attr_native_value = self._compute()
-        self.async_write_ha_state()
+        self._refresh_value()
 
 
 # ---------------------------------------------------------------------------
@@ -161,6 +177,7 @@ class DatapointsCountSensor(_DatapointsSensorBase):
     """Expose the total number of recorded datapoints."""
 
     _attr_icon = "mdi:counter"
+    _compute_blocks_io = True
 
     def __init__(self, entry: ConfigEntry, store: DatapointsStore) -> None:
         """Initialise the datapoint count sensor."""
@@ -178,6 +195,7 @@ class DatapointsLastTimestampSensor(_DatapointsSensorBase):
 
     _attr_device_class = SensorDeviceClass.TIMESTAMP
     _attr_icon = "mdi:clock-outline"
+    _compute_blocks_io = True
 
     def __init__(self, entry: ConfigEntry, store: DatapointsStore) -> None:
         """Initialise the last recorded timestamp sensor."""
@@ -206,6 +224,7 @@ class DatapointsLastMessageSensor(_DatapointsSensorBase):
     """Expose the message of the most recently recorded datapoint."""
 
     _attr_icon = "mdi:text"
+    _compute_blocks_io = True
 
     def __init__(self, entry: ConfigEntry, store: DatapointsStore) -> None:
         """Initialise the last recorded message sensor."""
@@ -228,6 +247,7 @@ class DatapointsTimeSinceLastSensor(_DatapointsPeriodicSensorBase):
     _attr_native_unit_of_measurement = UnitOfTime.HOURS
     _attr_suggested_display_precision = 1
     _attr_icon = "mdi:timer-outline"
+    _compute_blocks_io = True
 
     def __init__(
         self, entry: ConfigEntry, store: DatapointsStore, hass: HomeAssistant
@@ -259,6 +279,7 @@ class DatapointsTodayCountSensor(_DatapointsPeriodicSensorBase):
     """Expose the count of datapoints recorded since the start of today (local time)."""
 
     _attr_icon = "mdi:calendar-today"
+    _compute_blocks_io = True
 
     def __init__(
         self, entry: ConfigEntry, store: DatapointsStore, hass: HomeAssistant
@@ -278,6 +299,7 @@ class DatapointsWeekCountSensor(_DatapointsPeriodicSensorBase):
     """Expose the count of datapoints recorded since the start of this week (Mon, local time)."""
 
     _attr_icon = "mdi:calendar-week"
+    _compute_blocks_io = True
 
     def __init__(
         self, entry: ConfigEntry, store: DatapointsStore, hass: HomeAssistant
@@ -298,6 +320,7 @@ class DatapointsAutomationCountSensor(_DatapointsSensorBase):
     """Expose the count of automation-triggered datapoints."""
 
     _attr_icon = "mdi:robot"
+    _compute_blocks_io = True
 
     def __init__(self, entry: ConfigEntry, store: DatapointsStore) -> None:
         """Initialise the automation count sensor."""
@@ -315,6 +338,7 @@ class DatapointsManualCountSensor(_DatapointsSensorBase):
     """Expose the count of manually recorded datapoints."""
 
     _attr_icon = "mdi:hand-back-right"
+    _compute_blocks_io = True
 
     def __init__(self, entry: ConfigEntry, store: DatapointsStore) -> None:
         """Initialise the manual count sensor."""

custom_components/hass_datapoints/store.py

@@ -5,6 +5,7 @@
 import json
 import logging
 import sqlite3
+import threading
 import uuid
 from collections.abc import Callable
 from datetime import UTC, datetime
@@ -89,26 +90,36 @@ def _event_to_params(event: dict[str, Any]) -> tuple:
 
 
 class _EventDb:
-    """Synchronous SQLite backend for event storage."""
+    """Synchronous SQLite backend for event storage.
+
+    Holds a single long-lived connection guarded by a lock rather than opening
+    (and leaking) a fresh connection per operation. ``check_same_thread=False``
+    lets Home Assistant's executor threads reuse it; the lock serialises access
+    so the shared connection is never touched concurrently. All access is
+    blocking and MUST be dispatched from an executor, never the event loop.
+    """
 
     def __init__(self, db_path: str) -> None:
         self._db_path = db_path
+        self._lock = threading.Lock()
+        self._conn = sqlite3.connect(db_path, check_same_thread=False)
+        self._conn.row_factory = sqlite3.Row
         self._init_db()
 
-    def _connect(self) -> sqlite3.Connection:
-        conn = sqlite3.connect(self._db_path, check_same_thread=False)
-        conn.row_factory = sqlite3.Row
-        return conn
-
     def _init_db(self) -> None:
-        with self._connect() as conn:
-            conn.execute("PRAGMA journal_mode=WAL")
-            conn.executescript(_SCHEMA_SQL)
+        with self._lock, self._conn:
+            self._conn.execute("PRAGMA journal_mode=WAL")
+            self._conn.executescript(_SCHEMA_SQL)
+
+    def close(self) -> None:
+        """Close the underlying connection. Called once when the entry unloads."""
+        with self._lock:
+            self._conn.close()
 
     def insert(self, event: dict[str, Any]) -> None:
         """Insert a new event row."""
-        with self._connect() as conn:
-            conn.execute(
+        with self._lock, self._conn:
+            self._conn.execute(
                 """
                 INSERT INTO events
                     (id, timestamp, message, annotation,
@@ -122,8 +133,8 @@ def insert(self, event: dict[str, Any]) -> None:
     def insert_many(self, events: list[dict[str, Any]]) -> None:
         """Bulk-insert events using INSERT OR IGNORE (idempotent for migration)."""
         params = [_event_to_params(e) for e in events]
-        with self._connect() as conn:
-            conn.executemany(
+        with self._lock, self._conn:
+            self._conn.executemany(
                 """
                 INSERT OR IGNORE INTO events
                     (id, timestamp, message, annotation,
@@ -141,20 +152,20 @@ def update(self, event_id: str, fields: dict[str, Any]) -> bool:
         }
         cols = ", ".join(f"{k}=?" for k in encoded)
         params = list(encoded.values()) + [event_id]
-        with self._connect() as conn:
-            cursor = conn.execute(f"UPDATE events SET {cols} WHERE id=?", params)
+        with self._lock, self._conn:
+            cursor = self._conn.execute(f"UPDATE events SET {cols} WHERE id=?", params)
             return cursor.rowcount > 0
 
     def delete(self, event_id: str) -> bool:
         """Delete an event by ID. Returns True if the row existed."""
-        with self._connect() as conn:
-            cursor = conn.execute("DELETE FROM events WHERE id=?", (event_id,))
+        with self._lock, self._conn:
+            cursor = self._conn.execute("DELETE FROM events WHERE id=?", (event_id,))
             return cursor.rowcount > 0
 
     def delete_dev(self) -> int:
         """Delete all dev-flagged events. Returns the number of rows deleted."""
-        with self._connect() as conn:
-            cursor = conn.execute("DELETE FROM events WHERE dev=1")
+        with self._lock, self._conn:
+            cursor = self._conn.execute("DELETE FROM events WHERE dev=1")
             return cursor.rowcount
 
     def query(
@@ -165,17 +176,26 @@ def query(
         limit: int | None,
         offset: int,
     ) -> list[dict[str, Any]]:
-        """Return events filtered by time range, then by entity IDs in Python."""
-        with self._connect() as conn:
-            rows = conn.execute(
-                """
-                SELECT * FROM events
-                WHERE (? IS NULL OR timestamp >= ?)
-                AND   (? IS NULL OR timestamp <= ?)
-                ORDER BY timestamp ASC
-                """,
-                (start, start, end, end),
-            ).fetchall()
+        """Return events filtered by time range (and, in Python, by entity IDs).
+
+        With no entity filter, LIMIT/OFFSET are pushed into SQL so the whole
+        table is never materialised. With an entity filter the JSON-array
+        intersection has to run in Python, so pagination is applied after it.
+        """
+        sql = [
+            "SELECT * FROM events",
+            "WHERE (? IS NULL OR timestamp >= ?)",
+            "AND   (? IS NULL OR timestamp <= ?)",
+            "ORDER BY timestamp ASC",
+        ]
+        params: list[Any] = [start, start, end, end]
+        push_pagination = entity_ids is None
+        if push_pagination and (limit is not None or offset):
+            # SQLite uses LIMIT -1 to mean "no limit" when only an offset is set.
+            sql.append("LIMIT ? OFFSET ?")
+            params.extend([limit if limit is not None else -1, offset])
+        with self._lock:
+            rows = self._conn.execute("\n".join(sql), params).fetchall()
 
         events = [_row_to_dict(r) for r in rows]
 
@@ -191,49 +211,48 @@ def query(
                     seen.add(ev["id"])
                     filtered.append(ev)
             events = filtered
-
-        if offset:
-            events = events[offset:]
-        if limit is not None:
-            events = events[:limit]
+            if offset:
+                events = events[offset:]
+            if limit is not None:
+                events = events[:limit]
 
         return events
 
     def count(self) -> int:
         """Return total event count."""
-        with self._connect() as conn:
-            row = conn.execute("SELECT COUNT(*) FROM events").fetchone()
+        with self._lock:
+            row = self._conn.execute("SELECT COUNT(*) FROM events").fetchone()
             return row[0]
 
     def count_in_range(self, start: str, end: str | None) -> int:
         """Return event count within a time range."""
-        with self._connect() as conn:
-            row = conn.execute(
+        with self._lock:
+            row = self._conn.execute(
                 "SELECT COUNT(*) FROM events WHERE timestamp >= ? AND (? IS NULL OR timestamp <= ?)",
                 (start, end, end),
             ).fetchone()
             return row[0]
 
     def bounds(self) -> tuple[str | None, str | None]:
         """Return (earliest_timestamp, latest_timestamp) or (None, None) if empty."""
-        with self._connect() as conn:
-            row = conn.execute(
+        with self._lock:
+            row = self._conn.execute(
                 "SELECT MIN(timestamp), MAX(timestamp) FROM events"
             ).fetchone()
             return row[0], row[1]
 
     def last(self) -> dict[str, Any] | None:
         """Return the most recently timestamped event, or None if empty."""
-        with self._connect() as conn:
-            row = conn.execute(
+        with self._lock:
+            row = self._conn.execute(
                 "SELECT * FROM events ORDER BY timestamp DESC LIMIT 1"
             ).fetchone()
             return _row_to_dict(row) if row else None
 
     def automation_manual_counts(self) -> tuple[int, int]:
         """Return (automation_count, manual_count)."""
-        with self._connect() as conn:
-            row = conn.execute(
+        with self._lock:
+            row = self._conn.execute(
                 """
                 SELECT
                     COUNT(CASE WHEN automation_id IS NOT NULL THEN 1 END),
@@ -245,8 +264,8 @@ def automation_manual_counts(self) -> tuple[int, int]:
 
     def get_by_id(self, event_id: str) -> dict[str, Any] | None:
         """Fetch a single event by ID, or None if not found."""
-        with self._connect() as conn:
-            row = conn.execute(
+        with self._lock:
+            row = self._conn.execute(
                 "SELECT * FROM events WHERE id=?", (event_id,)
             ).fetchone()
             return _row_to_dict(row) if row else None
@@ -301,6 +320,10 @@ async def async_load(self) -> None:
             if "last_resolved_clusters_summary" not in m:
                 m["last_resolved_clusters_summary"] = []
 
+    def close(self) -> None:
+        """Release the SQLite connection. Called when the config entry unloads."""
+        self._event_db.close()
+
     async def async_record(
         self,
         message: str,

custom_components/hass_datapoints/websocket_api.py

@@ -190,12 +190,14 @@ async def ws_get_events(
         ]
     limit: int = msg.get("limit", 200)
     offset: int = msg.get("offset", 0)
-    events = store.get_events(
-        start=msg.get("start_time"),
-        end=msg.get("end_time"),
-        entity_ids=entity_ids,
-        limit=limit,
-        offset=offset,
+    # SQLite reads block, so run them in the executor rather than on the loop.
+    events = await hass.async_add_executor_job(
+        store.get_events,
+        msg.get("start_time"),
+        msg.get("end_time"),
+        entity_ids,
+        limit,
+        offset,
     )
     connection.send_result(msg["id"], {"events": events})
 
@@ -219,7 +221,9 @@ async def ws_get_event_bounds(
             _get_global_history_bounds, recorder
         )
         if start_time is None and end_time is None:
-            start_time, end_time = store.get_event_bounds()
+            start_time, end_time = await hass.async_add_executor_job(
+                store.get_event_bounds
+            )
             source = "datapoints_store_fallback"
         connection.send_result(
             msg["id"],

package.json

@@ -1,6 +1,6 @@
 {
   "name": "hass-datapoints",
-  "version": "0.6.3",
+  "version": "0.6.4",
   "private": true,
   "description": "Build and development entrypoints for the hass-datapoints Home Assistant integration.",
   "packageManager": "pnpm@10.33.0",