koa_middleware.database.local_database

Classes

LocalCalibrationDB(db_path, table_name)

Class to interact with a local SQLite calibration database using sqlite-utils.
class koa_middleware.database.local_database.LocalCalibrationDB(db_path: str, table_name: str)[source]

Bases: object

Class to interact with a local SQLite calibration database using sqlite-utils.

This class provides a simple interface for adding, querying, and managing calibration metadata stored as dictionaries in a SQLite database.

add(cals: dict | Sequence[dict], alter: bool = True)[source]

Add or update calibration entries in the database.

Parameters:

  • cals (dict | Sequence[dict]) – A single calibration metadata dictionary or a sequence of calibration metadata dictionaries to add or update. Uses upsert semantics with ‘id’ as primary key.

  • alter (bool, optional) – Whether to automatically alter the table schema to accommodate new fields. Default is True.

close()[source]

Close the database connection.

custom_query(sql: str, params: tuple = ()) → list[dict][source]

Execute a custom SQL query.

Parameters:

  • sql (str) – The SQL query string.

  • params (tuple, optional) – Parameters to pass to the SQL query.

Returns:

List of matching rows as dictionaries.

Return type:

list[dict]

delete(cal_id: str)[source]

Delete a calibration entry by its unique ID.

Parameters:

cal_id (str) – The unique calibration ID (UUID) to delete.

get_column(column: str) → list[dict][source]
get_last_updated() → str | None[source]

Get the most recent last_updated timestamp from the database.

Returns:

The maximum last_updated value as a string, or None if the table is empty.

Return type:

str | None

query(filename: str | None = None, cal_type: str | None = None, cal_id: str | None = None, cal_version_min: str | None = None, cal_version_max: str | None = None, date_time_start: str | None = None, date_time_end: str | None = None, last_updated_start: str | None = None, last_updated_end: str | None = None, origin: str | None = None, order_by: str = 'last_updated', fetch: str = 'all') → list[dict] | dict | None[source]

Query calibration entries from the database with common use cases.

Parameters:

  • filename (str, optional) – Filter by calibration filename.

  • cal_type (str, optional) – Filter by calibration type.

  • cal_id (str, optional) – Filter by a specific calibration UUID. Overrides other filters if provided.

  • cal_version_min (str, optional) – Minimum calibration version to include. Default is “001”

  • cal_version_max (str, optional) – Maximum calibration version to include. Default is “999”

  • date_time_start (str, optional) – Minimum datetime_obs to include.

  • date_time_end (str, optional) – Maximum datetime_obs to include.

  • last_updated_start (str, optional) – Minimum last_updated timestamp to include.

  • last_updated_end (str, optional) – Maximum last_updated timestamp to include.

  • origin (str, optional) – Filter by origin (“ANY”, “LOCAL”, “REMOTE”). Default is None (equivalent to “ANY”). Whether to return all matching rows or just the first one.

Returns:

Matching calibration entries. If fetch=’first’, returns a single dict or None. If fetch=’all’, returns a list of dicts.

Return type:

list[dict] or dict or None

query_filename(filename: str) → dict | None[source]

Query a calibration entry by its unique ID.

Parameters:

filename (str) – The unique calibration filename.

Returns:

The calibration metadata dictionary if found, otherwise None.

Return type:

dict or None

query_id(cal_id: str) → dict | None[source]

Query a calibration entry by its unique ID.

Parameters:

cal_id (str) – The unique calibration ID (UUID).

Returns:

The calibration metadata dictionary if found, otherwise None.

Return type:

dict or None

property rows: list[dict]

Get all rows in the calibration table.

Returns:

Generator of all calibration entries as dictionaries. Call list() on the result to get a list.

Return type:

Generator[dict]

property rows_where

Forward function to sqlite-utils Table.rows_where method.

property table

Returns the calibration table object.

Returns:

The table object for the calibration metadata.

Return type:

sqlite_utils.db.Table

transaction()[source]

Context manager for database transactions.

Ensures that changes are committed on success or rolled back on error.