Add Estimated Database Size to the recorder system health

homeassistant/components/recorder/const.py

@@ -37,3 +37,7 @@
 
 
 EXCLUDE_ATTRIBUTES = f"{DOMAIN}_exclude_attributes_by_domain"
+
+DIALECT_SQLITE = "sqlite"
+DIALECT_MYSQL = "mysql"
+DIALECT_POSTGRESQL = "postgresql"

homeassistant/components/recorder/core.py

@@ -193,6 +193,11 @@ def backlog(self) -> int:
         """Return the number of items in the recorder backlog."""
         return self._queue.qsize()
 
+    @property
+    def dialect(self) -> str | None:
+        """Return the dialect the recorder uses."""
+        return self.engine.dialect.name if self.engine else None
+
     @property
     def _using_file_sqlite(self) -> bool:
         """Short version to check if we are using sqlite3 as a file."""
@@ -462,7 +467,7 @@ def async_external_statistics(
     @callback
     def using_sqlite(self) -> bool:
         """Return if recorder uses sqlite as the engine."""
-        return bool(self.engine and self.engine.dialect.name == "sqlite")
+        return bool(self.dialect == "sqlite")
 
     @callback
     def _async_setup_periodic_tasks(self) -> None:

homeassistant/components/recorder/strings.json

@@ -2,7 +2,8 @@
   "system_health": {
     "info": {
       "oldest_recorder_run": "Oldest Run Start Time",
-      "current_recorder_run": "Current Run Start Time"
+      "current_recorder_run": "Current Run Start Time",
+      "estimated_db_size": "Estimated Database Size (MiB)"
     }
   }
 }

homeassistant/components/recorder/system_health/__init__.py

@@ -0,0 +1,59 @@
+"""Provide info to system health."""
+from __future__ import annotations
+
+from collections.abc import Callable
+from typing import Any
+
+from sqlalchemy.orm.session import Session
+from yarl import URL
+
+from homeassistant.components import system_health
+from homeassistant.core import HomeAssistant, callback
+
+from .. import get_instance
+from ..const import DIALECT_MYSQL, DIALECT_POSTGRESQL, DIALECT_SQLITE
+from .mysql import db_size_bytes as mysql_db_size_bytes
+from .postgresql import db_size_bytes as postgresql_db_size_bytes
+from .sqlite import db_size_bytes as sqlite_db_size_bytes
+
+DIALECT_TO_GET_SIZE = {
+    DIALECT_SQLITE: sqlite_db_size_bytes,
+    DIALECT_MYSQL: mysql_db_size_bytes,
+    DIALECT_POSTGRESQL: postgresql_db_size_bytes,
+}
+
+
+@callback
+def async_register(
+    hass: HomeAssistant, register: system_health.SystemHealthRegistration
+) -> None:
+    """Register system health callbacks."""
+    register.async_register_info(system_health_info)
+
+
+def _get_estimated_db_size(
+    session_maker: Callable[[], Session], dialect: str, database_name: str
+) -> str | None:
+    """Get the estimated size of the database."""
+    if (get_size := DIALECT_TO_GET_SIZE.get(dialect)) and (
+        db_bytes := get_size(session_maker(), database_name)
+    ):
+        return f"{db_bytes/1024/1024:.2f} MiB"
+    return None
+
+
+async def system_health_info(hass: HomeAssistant) -> dict[str, Any]:
+    """Get info for the info page."""
+    instance = get_instance(hass)
+    run_history = instance.run_history
+    database_name = URL(instance.db_url).path.lstrip("/")
+    estimated_db_size = await instance.async_add_executor_job(
+        _get_estimated_db_size, instance.get_session, instance.dialect, database_name
+    )
+    health_data = {
+        "oldest_recorder_run": run_history.first.start,
+        "current_recorder_run": run_history.current.start,
+    }
+    if estimated_db_size:
+        health_data["estimated_db_size"] = estimated_db_size
+    return health_data

homeassistant/components/recorder/system_health/mysql.py

@@ -0,0 +1,19 @@
+"""Provide info to system health for mysql."""
+from __future__ import annotations
+
+from sqlalchemy import text
+from sqlalchemy.orm.session import Session
+
+
+def db_size_bytes(session: Session, database_name: str) -> float:
+    """Get the mysql database size."""
+    return float(
+        session.execute(
+            text(
+                "SELECT ROUND(SUM(DATA_LENGTH + INDEX_LENGTH), 2) "
+                "FROM information_schema.TABLES WHERE "
+                "TABLE_SCHEMA=:database_name"
+            ),
+            {"database_name": database_name},
+        ).first()[0]
+    )

homeassistant/components/recorder/system_health/postgresql.py

@@ -0,0 +1,15 @@
+"""Provide info to system health for postgresql."""
+from __future__ import annotations
+
+from sqlalchemy import text
+from sqlalchemy.orm.session import Session
+
+
+def db_size_bytes(session: Session, database_name: str) -> float:
+    """Get the mysql database size."""
+    return float(
+        session.execute(
+            text("select pg_database_size(:database_name);"),
+            {"database_name": database_name},
+        ).first()[0]
+    )

homeassistant/components/recorder/system_health/sqlite.py

@@ -0,0 +1,17 @@
+"""Provide info to system health for sqlite."""
+from __future__ import annotations
+
+from sqlalchemy import text
+from sqlalchemy.orm.session import Session
+
+
+def db_size_bytes(session: Session, database_name: str) -> float:
+    """Get the mysql database size."""
+    return float(
+        session.execute(
+            text(
+                "SELECT page_count * page_size as size "
+                "FROM pragma_page_count(), pragma_page_size();"
+            )
+        ).first()[0]
+    )

homeassistant/components/recorder/translations/en.json

@@ -2,7 +2,8 @@
     "system_health": {
         "info": {
             "current_recorder_run": "Current Run Start Time",
-            "oldest_recorder_run": "Oldest Run Start Time"
+            "oldest_recorder_run": "Oldest Run Start Time",
+            "estimated_db_size": "Estimated Database Size"
         }
     }
 }