database

dev_tool.db.sqlite.database

log = logging.getLogger(__name__) module-attribute

SQLiteDatabase

A class for SQLite database operations.

This class provides methods for managing SQLite database connections and performing CRUD operations on destruction schedules.

The constructor for the SQLiteDatabase class.

Parameters:

• path (Path | None, default: None ) –

  The path to the SQLite database file, defaults to application database.

Source code in dev_tool/db/sqlite/database.py

def __init__(self, path: Path | None = None) -> None:
    """
    The constructor for the SQLiteDatabase class.

    :param path: The path to the SQLite database file, defaults to application database.
    """

    self.path = path or STRATUS / 'dev_tool.db'

path = path or STRATUS / 'dev_tool.db' instance-attribute

get_connection

A context manager that provides a SQLite database connection.

:yield: A SQLite database connection.

Source code in dev_tool/db/sqlite/database.py

@contextmanager
def get_connection(self) -> Generator[sqlite3.Connection, None, None]:
    """
    A context manager that provides a SQLite database connection.

    :yield: A SQLite database connection.
    """

    connection = None

    try:
        connection = self._connect()
        yield connection
    finally:
        if connection:
            connection.close()

add_schedule

A method that adds or updates a destruction schedule.

Parameters:

• name (str) –

  The name of the schedule entry.

• destructing_at (float) –

  The timestamp when destruction should occur.

Raises:

• SQLiteConnectionError –

  If schedule addition fails.

Source code in dev_tool/db/sqlite/database.py

def add_schedule(self, name: str, destructing_at: float) -> None:
    """
    A method that adds or updates a destruction schedule.

    :param name: The name of the schedule entry.
    :param destructing_at: The timestamp when destruction should occur.
    :raises SQLiteConnectionError: If schedule addition fails.
    """

    try:
        with self.get_connection() as connection:
            query = """
                INSERT OR REPLACE INTO destruction_schedule (name, destructing_at, created_at)
                VALUES (?, ?, ?)
            """

            connection.execute(
                query,
                (name, destructing_at, time.time())
            )

            connection.commit()
    except SQLiteConnectionError:
        raise
    except Exception as exception:
        message = f'Failed to add schedule for {name}: {exception}'
        CONTEXT.notification.error_banner(message)

        log.exception(message)
        raise SQLiteConnectionError(message) from None

get_schedules

A method that retrieves all destruction schedules.

Returns:

• list[dict[str, Any]] –

  A list of schedule dictionaries containing name and destructing_at.

Raises:

• SQLiteConnectionError –

  If schedule retrieval fails.

Source code in dev_tool/db/sqlite/database.py

def get_schedules(self) -> list[dict[str, Any]]:
    """
    A method that retrieves all destruction schedules.

    :return: A list of schedule dictionaries containing name and destructing_at.
    :raises SQLiteConnectionError: If schedule retrieval fails.
    """

    try:
        with self.get_connection() as connection:
            query = 'SELECT name, destructing_at FROM destruction_schedule'
            cursor = connection.execute(query)
            return [dict(row) for row in cursor.fetchall()]
    except SQLiteConnectionError:
        raise
    except Exception as exception:
        message = f'Failed to get schedules: {exception}'
        CONTEXT.notification.error_banner(message)

        log.exception(message)
        raise SQLiteConnectionError(message) from None

remove_schedule

A method that removes a destruction schedule by name.

Parameters:

• name (str) –

  The name of the schedule to remove.

Raises:

• SQLiteConnectionError –

  If schedule removal fails.

Source code in dev_tool/db/sqlite/database.py

def remove_schedule(self, name: str) -> None:
    """
    A method that removes a destruction schedule by name.

    :param name: The name of the schedule to remove.
    :raises SQLiteConnectionError: If schedule removal fails.
    """

    try:
        with self.get_connection() as connection:
            query = 'DELETE FROM destruction_schedule WHERE name = ?'
            connection.execute(query, (name,))
            connection.commit()
    except SQLiteConnectionError:
        raise
    except Exception as exception:
        message = f'Failed to remove schedule for {name}: {exception}'
        CONTEXT.notification.error_banner(message)

        log.exception(message)
        raise SQLiteConnectionError(message) from None

time_remaining

A method that calculates the remaining time before destruction.

Parameters:

• name (str) –

  The name of the schedule to check.

Returns:

• float | None –

  The remaining time in seconds, or None if schedule doesn't exist or expired.

Raises:

• SQLiteConnectionError –

  If time calculation fails.

Source code in dev_tool/db/sqlite/database.py

def time_remaining(self, name: str) -> float | None:
    """
    A method that calculates the remaining time before destruction.

    :param name: The name of the schedule to check.
    :return: The remaining time in seconds, or None if schedule doesn't exist or expired.
    :raises SQLiteConnectionError: If time calculation fails.
    """

    try:
        with self.get_connection() as connection:
            query = 'SELECT destructing_at FROM destruction_schedule WHERE name = ?'
            cursor = connection.execute(query, (name,))
            result = cursor.fetchone()

        if not result:
            return None

        remaining = result['destructing_at'] - time.time()
    except SQLiteConnectionError:
        raise
    except Exception as exception:
        message = f'Failed to get time remaining for {name}: {exception}'
        CONTEXT.notification.error_banner(message)

        log.exception(message)
        raise SQLiteConnectionError(message) from None
    else:
        return remaining if remaining > 0 else None