bhf_smarthealth.load.sync_db

bhf_smarthealth.load.sync_db#

This module contains utility functions for connecting to, and updating an in-process DuckDB database instance

Attributes#

logger

Functions#

load_target_as_table(→ None)

Connects to DuckDB using an in-memory database and loads the target Parquet

get_upsert_counts(→ Dict[str, int])

Calculates UPSERT counts (updated and inserted rows) using a single, efficient

run_upsert_update(→ None)

Performs an UPSERT (Update or Insert) operation on a Parquet file

save_table_to_parquet(conn, table_name, output_path)

Save a specified table to a parquet file

execute_upsert(table_name, target_parquet_path, ...)

Overarching function for executing the upsert workflow

Module Contents#

bhf_smarthealth.load.sync_db.logger#
bhf_smarthealth.load.sync_db.load_target_as_table(conn: duckdb.DuckDBPyConnection, table_name: str, target_parquet_path: os.PathLike) → None[source]#

Connects to DuckDB using an in-memory database and loads the target Parquet file into a temporary, mutable table with the given name.

Parameters:

  • conn (duckdb.DuckDBPyConnection) – An active DuckDB connection object.

  • table_name (str) – The name of the in-memory table to create.

  • target_parquet_path (str) – The path to the target Parquet file to load.

bhf_smarthealth.load.sync_db.get_upsert_counts(conn: duckdb.DuckDBPyConnection, target_table_name: str, update_parquet_path: os.PathLike, join_conditions: str, key_column: str) → Dict[str, int][source]#

Calculates UPSERT counts (updated and inserted rows) using a single, efficient LEFT JOIN query on the update file and the registered target table.

Parameters:

  • conn (duckdb.DuckDBPyConnection) – An active DuckDB connection object.

  • target_table_name (str) – The name of the temporary DuckDB table containing the target data (t).

  • update_parquet_path (str) – The file path of the update Parquet file (u).

  • key_column (str) – Column name that is used to check matching rows.

Returns:

A dictionary containing ‘rows_updated’, ‘rows_inserted’, and ‘total_incoming’ on success, or None if an exception occurs during calculation.

Return type:

Dict[str, int]

bhf_smarthealth.load.sync_db.run_upsert_update(conn: duckdb.DuckDBPyConnection, table_name: str, update_parquet_path: os.PathLike, join_conditions: str) → None[source]#

Performs an UPSERT (Update or Insert) operation on a Parquet file using an in-memory DuckDB instance via the efficient MERGE INTO statement.

The function loads the TARGET file into an in-memory table, merges the data from the UPDATE file, and then overwrites the original TARGET file with the final, reconciled table state.

Parameters:

  • conn (duckdb.DuckDBPyConnection) – An active DuckDB connection object.

  • table_name (str) – The name of the temporary in-memory table to update (e.g., “intraday_activities”).

  • update_parquet_path (str) – The file path to the Parquet file containing the new/update rows.

  • join_conditions (str) – SQL snippet specifying the columns to match on

Returns:

The function modifies the TARGET Parquet file in place (via overwrite) and returns no explicit value on successful execution.

Return type:

None

bhf_smarthealth.load.sync_db.save_table_to_parquet(conn: duckdb.DuckDBPyConnection, table_name: str, output_path: os.PathLike)[source]#

Save a specified table to a parquet file

Parameters:

  • conn (duckdb.DuckDBPyConnection) – An active DuckDB connection object.

  • table_name (str) – The name of the temporary in-memory table to save

  • output_path (PathLike) – Path to save the table to

bhf_smarthealth.load.sync_db.execute_upsert(table_name: str, target_parquet_path: os.PathLike, update_parquet_path: os.PathLike, temp_output_path: os.PathLike, key_columns: List[str])[source]#

Overarching function for executing the upsert workflow

Parameters:

  • table_name (str) – The name of the temporary in-memory table to load the target parquet file into

  • target_parquet_path (PathLike) – Path to the target parquet file

  • update_parquet_path (PathLike) – Path to the update parquet file

  • temp_output_path (PathLike) – Path to the temp output parquet file

  • key_columns (List[str]) – Key columns to join the tables on

On this page