koa_middleware.database.metadata_database

Classes

CalibrationDB(url, orm_class)

Generic utility class to interface with a local SQLite DB or remote PostgreSQL DB.
class koa_middleware.database.metadata_database.CalibrationDB(url: str, orm_class: type[CalibrationORM])[source]

Bases: object

Generic utility class to interface with a local SQLite DB or remote PostgreSQL DB.

add(calibration: CalibrationORM | list[CalibrationORM], session: Session | None = None, commit: bool = True)[source]

Add one or many calibrations to the database.

Parameters:

  • calibration (CalibrationORM | list[CalibrationORM]) – A single CalibrationORM object or a list of them to add.

  • session (Session | None) – Optional SQLAlchemy session to use. If None, a new session will be created.

  • commit (bool) – Whether to commit the transaction after adding the calibration(s). Defaults to True.

close()[source]
get_engine(url: str, echo: bool = True)[source]
get_last_updated(session: Session | None = None) → str[source]

Get most recent LAST_UPDATED timestamp from the database.

Parameters:

session (Session | None) – Optional SQLAlchemy session to use. If None, a new session will be created.

Returns:

str – The most recent LAST_UPDATED timestamp in ISO format.

query(session: Session | None = None, cal_type: str | None = None, date_time_start: str | None = None, date_time_end: str | None = None, fetch: str = 'all') → list[CalibrationORM][source]

Higher level query method to retrieve calibrations based on a specified calibration type and datetime start/end. The utility of this method will be revisited as the DRP is developed.

Parameters:

  • session (Session | None) – Optional SQLAlchemy session to use. If None, a new session will be created.

  • cal_type (str | None) – Optional calibration type to filter results. If None, all types are included.

  • date_time_start (str | None) – Start datetime in ISO format. Defaults to the minimum datetime.

  • date_time_end (str | None) – End datetime in ISO format. Defaults to the maximum datetime.

  • fetch (str) – Specifies whether to fetch ‘all’ results or just the ‘first’ result. Defaults to ‘all’.

Returns:

list[CalibrationORM] – A list of CalibrationORM objects matching the query criteria.

query_by_id(calibration_id: str, session: Session | None = None) → CalibrationORM | None[source]

Query a calibration by its ID.

Parameters:

  • calibration_id (str) – The ID of the calibration to query.

  • session (Session | None) – Optional SQLAlchemy session to use. If None, a new session will be created.

Returns:

list[CalibrationORM] | None – The calibration objects found by ID.

session_manager(external_session: Session | None = None)[source]

Context manager to handle database sessions.