flexmeasures.data.services.data_ingestion

Logic around data ingestion (jobs)

Functions

flexmeasures.data.services.data_ingestion.add_beliefs_to_db_and_enqueue_forecasting_jobs(data: BeliefsDataFrame | list[BeliefsDataFrame] | None = None, sensor_id: int | None = None, user_id: int | None = None, sensor_data: dict | None = None, uploaded_files: list[dict] | None = None, upload_data: dict | None = None, forecasting_jobs: list[Job] | None = None, forecasting_job_ids: list[str] | None = None, save_changed_beliefs_only: bool = True) → str

Save sensor data to the database and optionally enqueue forecasting jobs.

This function is intended to be called as an RQ job by an ingestion queue worker, but can also be called directly (e.g. as a fallback when no workers are available).

Parameters:

  • data – BeliefsDataFrame (or list thereof) to be saved.

  • sensor_id – Sensor ID for raw JSON or file ingestion.

  • user_id – User ID used to resolve the source of raw ingested data.

  • sensor_data – Raw JSON payload from the sensor data endpoint.

  • uploaded_files – Uploaded file contents and metadata.

  • upload_data – Raw form payload from the sensor data upload endpoint.

  • forecasting_jobs – Optional list of forecasting Jobs to enqueue after saving.

  • forecasting_job_ids – Optional list of forecasting Job ids to enqueue after saving.

  • save_changed_beliefs_only – If True, skip saving beliefs whose value hasn’t changed.

Returns:

Status string as returned by save_to_db.

Raises:

BeliefReplacementNotAllowed – if saving would change existing beliefs, while FLEXMEASURES_ALLOW_DATA_OVERWRITE is off.

flexmeasures.data.services.data_ingestion.is_belief_replacement(error: IntegrityError) → bool

Tell whether an IntegrityError is a UniqueViolation on the TimedBelief primary key.

The error is recognised by its SQLSTATE rather than by its class, because the class depends on the database driver: psycopg2 exposes the SQLSTATE as pgcode, and psycopg 3 as sqlstate.

Exceptions

exception flexmeasures.data.services.data_ingestion.BeliefReplacementNotAllowed(error: IntegrityError)

Saving would change existing beliefs, which this server does not allow.

It is the IntegrityError that refused the replacement, renamed and with a readable message, so code catching IntegrityError around saving beliefs still catches it.