koa_middleware.database.metadata_database

Classes

CalibrationDB(url, orm_class)

A generic utility class for interfacing with calibration databases.

class koa_middleware.database.metadata_database.CalibrationDB(url: str, orm_class: type[CalibrationORM])[source]

Bases: object

A generic utility class for interfacing with calibration databases.

This class provides a foundational layer for database operations, supporting both local SQLite and remote PostgreSQL databases. It manages SQLAlchemy engine creation, session management, and common CRUD (Create, Read, Update, Delete) operations for CalibrationORM objects.

It is designed to be extended by specific database implementations (e.g., LocalCalibrationDB, RemoteCalibrationDB).

engine

The SQLAlchemy engine connected to the database.

Type:: sqlalchemy.engine.base.Engine

Session

A session factory for creating new database sessions.

Type:: sqlalchemy.orm.session.sessionmaker

orm_class

The SQLAlchemy ORM class used to define the database schema.

Type:: type[CalibrationORM]

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

Adds one or more calibration objects to the database.

This method uses session.merge() to add or update calibration entries, allowing for idempotent operations (i.e., adding an existing entry will update it).

Parameters:

calibration (CalibrationORM | list[CalibrationORM]) – A single CalibrationORM object or a list of CalibrationORM objects to add.
session (Session | None) – An optional SQLAlchemy session to use. If None, a new session will be created and managed.
commit (bool) – If True, the transaction will be committed after adding the calibration(s). Defaults to True. Set to False if you intend to perform more operations within the same session before committing.

Example

>>> # Assuming 'db' is an instance of CalibrationDB and 'new_cal' is a CalibrationORM object
>>> # db.add(new_cal)
>>> # db.add([cal1, cal2], commit=False) # Add multiple without immediate commit

close()[source]

Disposes of the SQLAlchemy engine, closing all connections in the connection pool.

It is important to call this method when the database object is no longer needed to release resources.

get_engine(url: str, echo: bool = True)[source]

Creates and returns a SQLAlchemy engine for the given database URL.

Parameters:

url (str) – The database connection URL.
echo (bool) – If True, SQLAlchemy will log all SQL statements to stdout. Defaults to True.

Returns:

sqlalchemy.engine.base.Engine – The created SQLAlchemy engine.

get_last_updated(session: Session | None = None) → str[source]

Retrieves the most recent LAST_UPDATED timestamp from the database.

This method queries the orm_class table to find the maximum value in the last_updated column.

Parameters:: session (Session | None) – An optional SQLAlchemy session to use. If None, a new session will be created and managed by the session_manager context.
Returns:: str – The most recent LAST_UPDATED timestamp as an ISO formatted string. Returns None if the table is empty or no last_updated values exist.

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]

Queries the database for calibration entries based on various criteria.

This method allows filtering calibrations by type and observation datetime range.

Parameters:

session (Session | None) – An optional SQLAlchemy session to use. If None, a new session will be created and managed.
cal_type (str | None) – An optional calibration type string to filter the results. If None, calibrations of all types are included.
date_time_start (str | None) – The start datetime (inclusive) for filtering calibrations, in ISO format (e.g., ‘YYYY-MM-DDTHH:MM:SS.sss’). If None, defaults to the earliest possible datetime.
date_time_end (str | None) – The end datetime (inclusive) for filtering calibrations, in ISO format (e.g., ‘YYYY-MM-DDTHH:MM:SS.sss’). If None, defaults to the latest possible datetime.
fetch (str) – Specifies how many results to fetch: - ‘all’: Fetches all matching results (default). - ‘first’: Fetches only the first matching result.

Returns:

list[CalibrationORM] – A list of CalibrationORM objects that match the query criteria. If fetch is ‘first’, the list will contain at most one element.

Example

>>> # Assuming 'db' is an instance of CalibrationDB
>>> # all_cals = db.query()
>>> # type_filtered_cals = db.query(cal_type='BIAS')
>>> # time_filtered_cals = db.query(date_time_start='2023-01-01T00:00:00', date_time_end='2023-01-31T23:59:59')
>>> # first_cal = db.query(fetch='first')

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

Queries the database for calibration entries by their unique ID.

Parameters:

calibration_id (str) – The unique identifier of the calibration to query.
session (Session | None) – An optional SQLAlchemy session to use. If None, a new session will be created and managed.

Returns:

list[CalibrationORM] – A list of CalibrationORM objects that match the given ID. Returns an empty list if no matching calibration is found.

Example

>>> # Assuming 'db' is an instance of CalibrationDB
>>> # cal = db.query_by_id('some_calibration_id')
>>> # if cal:
>>> #     print(f"Found calibration with ID: {cal[0].id}")

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

Provides a context manager for managing SQLAlchemy database sessions.

This context manager ensures that sessions are properly created, committed, rolled back on errors, and closed. It can either create its own session or use an externally provided session.

Parameters:: external_session (Session | None) – An optional existing SQLAlchemy session to use. If None, a new session will be created.
Yields:: Session – The SQLAlchemy session to be used within the context.
Raises:: SQLAlchemyError – Any exception raised during database operations within the context will cause a rollback and be re-raised.

Example

>>> # Assuming 'db' is an instance of CalibrationDB
>>> with db.session_manager() as session:
...     # Perform database operations using 'session'
...     pass
>>> # Or with an external session:
>>> # my_session = db.Session()
>>> # with db.session_manager(my_session) as session:
>>> #     pass
>>> # my_session.close()