astra.database_manager

astra.database_manager#

Classes

DatabaseManager(observatory_name[, ...])

Manages the SQLite database for an observatory, including creation, querying, and periodic backups.
class astra.database_manager.DatabaseManager(observatory_name: str, run_backup: bool = True, backup_time: datetime = datetime.strptime('12:00', '%H:%M'), logger=None)[source]#

Bases: object

Manages the SQLite database for an observatory, including creation, querying, and periodic backups.

observatory_name#

Name of the observatory.

Type:

str

run_backup#

Flag to indicate if a backup should be run.

Type:

bool

backup_time#

Scheduled time for daily backups.

Type:

datetime

db_path#

File path to the SQLite database.

Type:

str

logger#

Logger instance for logging messages.

Type:

ObservatoryLogger

Examples

>>> from astra.config import ObservatoryConfig
>>> from astra.database_manager import DatabaseManager
>>> observatory_config = ObservatoryConfig.from_config()
>>> db_manager = DatabaseManager(observatory_config.observatory_name)
>>> db_manager.execute_select_to_df("SELECT * FROM polling", table="polling")
property cursor: Sqlite3Worker#
execute(query: str)[source]#
execute_select(query: str) → list[tuple][source]#

Execute a SELECT query and return the result as a list of tuples. Only SELECT queries are allowed.

For static type checking to ensure the return type is always a list of tuples.

execute_select_to_df(query: str, table: str | None = None) → DataFrame[source]#

Execute a SELECT query and return the result as a pandas DataFrame. Only SELECT queries are allowed.

classmethod from_observatory_config(observatory_config)[source]#
create_database(max_queue_size: int = 2000) → Sqlite3Worker[source]#

Create and initialize the observatory database.

Creates a SQLite database for storing observatory data including device polling information, image metadata, and log entries. The database includes three main tables: polling (device status data), images (image file information), and log (system log messages).

Returns:

Sqlite3Worker – The database cursor object for executing queries and managing the database connection with a maximum queue size of 200.

Note

The database file is created in the logs directory using the observatory name as the filename with a .db extension.

backup() → None[source]#

Back up database tables from the previous 24 hours to CSV files.

Creates timestamped CSV backups of the main database tables (polling, log, autoguider_log, autoguider_info_log) and stores them in an archive directory. Also monitors disk usage and logs a warning if disk usage exceeds 90%.

The backup process:

  1. Checks available disk space and warns if usage > 90%

  2. Creates an archive directory if it doesn’t exist

  3. Exports specified database tables to timestamped CSV files

  4. Logs the backup completion or any errors encountered

Raises:

Exception – Any errors during the backup process are logged and added to the error_source list for monitoring.

is_now_backup_time() → bool[source]#

Check if the current time matches the scheduled backup time.

maybe_run_backup(thread_manager) → None[source]#

Check if it’s time to run a backup and, if so, start it in a separate thread. Appends the backup thread to the thread_manager.

Contents