repack.py

Go to the documentation of this file.

"""Purge repack helper."""
 
from __future__ import annotations
 
import logging
from typing import TYPE_CHECKING
 
from sqlalchemy import text
 
from .const import SupportedDialect
from .db_schema import ALL_TABLES
 
if TYPE_CHECKING:
    from . import Recorder
 
_LOGGER = logging.getLogger(__name__)
 
 
def repack_database(instance: Recorder) -> None:
    """Repack based on engine type."""
    assert instance.engine is not None
    dialect_name = instance.engine.dialect.name
 
    # Execute sqlite command to free up space on disk
    if dialect_name == SupportedDialect.SQLITE:
        _LOGGER.debug("Vacuuming SQL DB to free space")
        with instance.engine.connect() as conn:
            conn.execute(text("VACUUM"))
            conn.commit()
        return
 
    # Execute postgresql vacuum command to free up space on disk
    if dialect_name == SupportedDialect.POSTGRESQL:
        _LOGGER.debug("Vacuuming SQL DB to free space")
        with instance.engine.connect().execution_options(
            isolation_level="AUTOCOMMIT"
        ) as conn:
            conn.execute(text("VACUUM"))
            conn.commit()
        return
 
    # Optimize mysql / mariadb tables to free up space on disk
    if dialect_name == SupportedDialect.MYSQL:
        _LOGGER.debug("Optimizing SQL DB to free space")
        with instance.engine.connect() as conn:
            conn.execute(text(f"OPTIMIZE TABLE {','.join(ALL_TABLES)}"))
            conn.commit()
        return