koa_middleware.store

Classes

CalibrationStore(orm_class[, cache_dir, ...])

Manages the storage, retrieval, and synchronization of calibration data.
class koa_middleware.store.CalibrationStore(orm_class: type[CalibrationORM], cache_dir: str | None = None, local_database_filename: str | None = None, remote_database_url: str | None = None, calibrations_url: str | None = None, use_cached: bool | None = None)[source]

Bases: object

Manages the storage, retrieval, and synchronization of calibration data.

The CalibrationStore class provides a unified interface for interacting with both local (SQLite) and remote (PostgreSQL) calibration databases. It handles caching of calibration files, querying for specific calibrations, and synchronizing calibration metadata between local and remote repositories.

It relies on environment variables for default configurations, such as cache directory and database URLs, but these can be overridden during initialization.

orm_class

The SQLAlchemy ORM class used for database queries.

Type:

type[CalibrationORM]

use_cached

If True, cached calibrations are used; otherwise, calibrations are always downloaded.

Type:

bool

cache_dir

The base directory for storing cached calibration files and the local database.

Type:

str

calibrations_url

The URL from which calibration files can be downloaded.

Type:

str | None

local_db

An instance of the local SQLite database handler.

Type:

LocalCalibrationDB

remote_db

An instance of the remote PostgreSQL database handler, if configured.

Type:

RemoteCalibrationDB | None

calibration_in_cache(calibration: CalibrationORM) → str | None[source]

Checks if a calibration file is already present in the local cache.

Parameters:

calibration (CalibrationORM) – The ORM instance representing the calibration to check.

Returns:

str | None – The absolute local file path of the calibration if it exists in the cache, otherwise None.

close()[source]

Closes the connections to both local and remote databases.

This method calls engine.dispose() on the underlying SQLAlchemy engines for both self.local_db and self.remote_db (if they exist), releasing any held resources.

download_calibration(calibration: CalibrationORM) → str[source]

Downloads a calibration file from the remote URL.

This method is responsible for fetching the actual calibration file (e.g., FITS file) from the configured remote URL and storing it in the local cache directory.

Parameters:

calibration (CalibrationORM) – The ORM instance representing the calibration to download.

Returns:

str – The local file path of the downloaded calibration file.

Raises:

NotImplementedError – This method is currently under development and not yet implemented. It will be implemented once the remote KOA infrastructure is set up.

get_calibration(input, selector: CalibrationSelector, use_cached: bool | None = None, **kwargs) → tuple[CalibrationORM, str][source]

Selects the best calibration based on input data and a selection rule, then retrieves it.

This method uses a CalibrationSelector to identify the most appropriate calibration for the given input data. Once selected, it retrieves the calibration file, downloading it if it’s not already cached locally.

Parameters:

  • input – The input data product for which a calibration is needed. The type of this input depends on the specific selector used.

  • selector (CalibrationSelector) – An instance of a CalibrationSelector that defines the logic for selecting the best calibration from the database.

  • use_cached (bool | None) – If True, returns the cached calibration if available. If False, always downloads from the remote even if already cached. If None, defaults to self.use_cached.

  • **kwargs – Additional parameters to pass to the selector.select() method.

Returns:

tuple[CalibrationORM, str] – A tuple containing: - CalibrationORM: The ORM instance representing the selected calibration. - str: The local file path of the retrieved calibration file.

Example

>>> # Assuming 'my_input_data' and 'my_selector' are defined
>>> # local_filepath, calibration_orm = store.get_calibration(my_input_data, my_selector)
>>> # print(f"Calibration file: {local_filepath}")
>>> # print(f"Calibration ORM ID: {calibration_orm.id}")
get_calibration_by_id(calibration_id: str) → tuple[CalibrationORM | None, str | None][source]

Retrieves a calibration by its unique identifier.

This method queries the local database for a calibration matching the given ID. If found, it retrieves the calibration file (downloading if necessary).

Parameters:

calibration_id (str) – The unique identifier of the calibration to retrieve.

Returns:

tuple[CalibrationORM | None, str | None] – A tuple containing: - CalibrationORM | None: The ORM instance of the found calibration, or None if not found. - str | None: The local file path of the calibration file, or None if not found.

Warns:

  • UserWarning – If no calibration is found with the given ID.

  • UserWarning – If multiple calibrations are found with the same ID (returns the first one).

Example

>>> # local_filepath, calibration_orm = store.get_calibration_by_id('some_calibration_id')
>>> # if calibration_orm:
>>> #     print(f"Found calibration: {calibration_orm.id} at {local_filepath}")
>>> # else:
>>> #     print("Calibration not found.")
get_local_filepath(calibration: CalibrationORM) → str[source]

Constructs the expected local file path for a given calibration ORM object.

This method does not check for the existence of the file, only generates its path.

Parameters:

calibration (CalibrationORM) – The ORM instance representing the calibration.

Returns:

str – The absolute local file path where the calibration file is expected to be stored.

get_missing_local_entries() → list[CalibrationORM][source]

Identifies calibration entries present in the remote database but missing from the local database.

This method queries the remote database for entries that have been updated more recently than the last update recorded in the local database. It is intended to help synchronize the local cache with the remote source.

Returns:

list[CalibrationORM] – A list of CalibrationORM objects representing entries that are in the remote DB but not yet in the local DB.

Raises:

AttributeError – If self.remote_db is None (i.e., no remote database is configured).

Note

This method is currently under development and requires a formal remote DB configuration to be fully functional and tested.

init_cache(local_database_filename: str | None = None)[source]

Initializes the local calibration cache and database.

This method sets up the necessary directory structure for caching calibration files and initializes the LocalCalibrationDB instance for managing the local SQLite database.

Parameters:

local_database_filename (str | None) – The desired filename for the local SQLite database. If None, the method will attempt to use the environment variable KOA_LOCAL_DATABASE_FILENAME or generate a default filename based on the ORM class.

Side Effects:

  • Creates the cache_dir, cache_dir/calibrations, and cache_dir/database directories if they do not already exist.

  • Initializes self.local_db with a LocalCalibrationDB instance.

init_remote_db(remote_database_url: str | None = None)[source]

Initializes the connection to the remote calibration database.

This method attempts to establish a connection to a remote PostgreSQL database using the provided URL or an environment variable. If no URL is provided (either directly or via environment variable), the remote_db attribute will be set to None.

Parameters:

remote_database_url (str | None) – The URL for the remote PostgreSQL database. If None, the KOA_REMOTE_CALIBRATION_URL environment variable is checked.

Side Effects:

  • Initializes self.remote_db with a RemoteCalibrationDB instance if a URL is available, otherwise sets it to None.

register_local_calibration(calibration) → tuple[str, CalibrationORM][source]

Registers a calibration that has been saved to the local calibrations directory.

This method takes a calibration object (expected to be a data model with a save method) and adds its corresponding ORM instance to the local SQLite database.

Parameters:

calibration – The calibration object to register. This object is expected to have a save(output_dir) method that saves the calibration file and returns its local path, and a to_orm() method that converts it to a CalibrationORM instance.

Returns:

  • tuple[str, CalibrationORM] – A tuple containing:

  • - `str` – The local file path where the calibration was saved.

  • - `CalibrationORM` – The ORM instance representing the registered calibration, as added to the local database.

Note

This method assumes the input calibration object is a data model that handles its own saving to disk and conversion to an ORM object. Consider alternative approaches if this assumption changes.

sync_from_remote() → list[CalibrationORM][source]

Synchronizes the local database with the remote database.

This method fetches entries from the remote database that are missing from the local database (based on the LAST_UPDATED field) and adds them to the local database.

Returns:

list[CalibrationORM] – A list of CalibrationORM objects that were added to the local database during synchronization.

Note

This method is currently under development and requires a formal remote DB configuration to be fully functional and tested.