Skip to main content

pod_db_utils

Utilities for the Project database.

The project database is a per-pod SQLite file that acts as a results ledger for the "run on new data only" feature: it records which files a task has already processed (and their results) so subsequent runs can skip unchanged files. It is written with raw SQL / pandas to_sql rather than the SQLAlchemy cache layer.

It holds three kinds of table:

  • task_definitions: a static registry mapping each taskhash to the protocol and algorithm used. One row per unique task; inserts are idempotent.
  • "{task_hash}-v2": one dynamic table per task holding references to the files processed by that task, plus any selected result columns. The -v2 suffix is the schema version of this per-task table. The column schema comes from the datasource (get_project_db_sqlite_create_table_query); the filename column is the primary key with ON CONFLICT REPLACE, so re-processing a file overwrites its row.
  • "{task_hash}-failed-v1": one dynamic table per task holding references to the files that failed during that task, so they are not retried. -v1 is this table's schema version.

Module​

Functions​

get_failed_files_cache​

def get_failed_files_cache(    project_db_con: Connection, task_hash: str, datasource: FileSystemIterableSource,) ‑> dict[str, dict[str, str]]:

Retrieves failed files information from the project database.

Arguments

  • project_db_con: The connection to the project database.
  • task_hash: The hash of the task.
  • datasource: The datasource to get column info from.

Returns Dictionary mapping filename to failure info (error_message, failure_timestamp, last_modified)

map_task_to_hash_add_to_db​

def map_task_to_hash_add_to_db(    serialized_protocol: SerializedProtocol, task_hash: str, project_db_con: Connection,) ‑> None:

Maps the task hash to the protocol and algorithm used.

Adds the task to the task database if it is not already present.

Arguments

  • serialized_protocol: The serialized protocol used for the task.
  • task_hash: The hash of the task.
  • project_db_con: The connection to the database.

save_failed_files_to_project_db​

def save_failed_files_to_project_db(    project_db_con: Connection,    failed_files: dict[str, Exception],    task_hash: str,    datasource: FileSystemIterableSource,) ‑> None:

Saves failed files information to the project database.

save_processed_datapoint_to_project_db​

def save_processed_datapoint_to_project_db(    project_db_con: Connection,    datasource: FileSystemIterableSource,    task_hash: str,    results: pd.DataFrame | None = None,    save_columns: list[str] | None = None,) ‑> None:

Saves the result of a task run to the database.

Arguments

  • project_db_con: The connection to the project database.
  • datasource: The datasource used for the task.
  • task_hash: The hash of the task, a unique identifier for when results have come from the same task definition, regardless of whether they are from the same run.
  • results: Results from the run to save.
  • save_columns: The relevant columns from the results to save to cache.