pi_dashboard.db.metrics_database_manager

Metrics database manager.

Classes

MetricsDatabaseManager(db_config)	Manager class for metrics database operations.
SystemMetricsDB(*[, id])	System metrics table.

class pi_dashboard.db.metrics_database_manager.SystemMetricsDB(*, id=None, cpu_usage, memory_usage, disk_usage, uptime, temperature, timestamp)

System metrics table.

id: int | None

cpu_usage: float

memory_usage: float

disk_usage: float

uptime: int

temperature: float

timestamp: int

classmethod from_system_metrics(system_metrics)

Create a SystemMetricsDB instance from a SystemMetrics.

Return type:

SystemMetricsDB

to_system_metrics()

Convert the database model to a SystemMetrics.

Return type:

SystemMetrics

__init__(**data)

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

model_config: ClassVar[ConfigDict] = {'from_attributes': True, 'read_from_attributes': True, 'read_with_orm_mode': True, 'registry': PydanticUndefined, 'table': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pi_dashboard.db.metrics_database_manager.MetricsDatabaseManager(db_config)

Manager class for metrics database operations.

__init__(db_config)

Initialize the MetricsDatabaseManager with the given database configuration.

db_config: DashboardDatabaseConfig

property db_url: str

Get the database URL.

is_stale(entry)

Return True if the metrics entry is older than the specified metrics lifetime.

Parameters:

entry (SystemMetrics) – The metrics entry to check

Return bool:

True if the entry is stale

Return type:

bool

get_all_system_metrics_entries()

Public method to retrieve all metrics entries.

Return type:

list[SystemMetrics]

get_system_metrics_entries_since(last_n_seconds, max_data_points)

Get metrics entries from the last N seconds with adaptive downsampling.

Return type:

list[SystemMetrics]

perform_system_metrics_action(system_metrics, action)

Perform a metrics action (create/update/delete) on the database.

Return type:

int

cleanup_old_system_metrics()

Delete metrics entries that are older than the specified metrics lifetime.

Return type:

None