pain001.db package#

Database operations module for pain001.

Submodules#

pain001.db.load_db_data module#

Load payment data from SQLite databases.

pain001.db.load_db_data.load_db_data(data_file_path: str, table_name: str) → list[dict[str, Any]][source]#

Load data from an SQLite database table into a list of dictionaries.

Parameters:

  • data_file_path – The path to the SQLite database file.

  • table_name – The name of the table from which data will be loaded.

Returns:

A list of dictionaries where each dictionary represents a row of data. The keys in each dictionary correspond to the column names, and the values are the column values for that row.

Return type:

list

Raises:

FileNotFoundError – If the SQLite file specified by data_file_path does not exist or fails path validation. SQLite operational errors and ConfigurationError for invalid table names propagate unchanged.

Example

data = load_db_data(“my_database.db”, “my_table”)

pain001.db.load_db_data.sanitize_table_name(table_name: str) → str[source]#

Validate and sanitize a table name to prevent SQL injection. Uses strict validation: only alphanumeric characters and underscores allowed. MUST start with a letter (SQL identifier rules).

Parameters:

table_name – The table name to validate.

Returns:

The validated table name (unchanged if valid).

Return type:

str

Raises:

ConfigurationError – If the table name is empty or contains invalid characters.

pain001.db.load_db_data_streaming module#

Memory-efficient streaming loader for SQLite database.

pain001.db.load_db_data_streaming.load_db_data_streaming(data_file_path: str, table_name: str, chunk_size: int = 1000) → Generator[list[dict[str, Any]], None, None][source]#

Load data from SQLite database in chunks for memory-efficient processing.

This function yields chunks of database rows instead of loading the entire table into memory, making it suitable for large databases.

Parameters:

  • data_file_path – The path to the SQLite database file.

  • table_name – The name of the table from which data will be loaded.

  • chunk_size – Number of rows to yield per chunk. Default is 1000.

Yields:

list[dict[str, Any]] – Up to chunk_size rows of table data.

Raises:

  • FileNotFoundError – If the SQLite file does not exist or fails path validation.

  • DataSourceError – If the table is empty or doesn’t exist.

Example

>>> for chunk in load_db_data_streaming('payments.db', 'pain001', chunk_size=500):
...     # Process chunk
...     process_payment_batch(chunk)
Performance:

  • Memory usage: O(chunk_size) instead of O(table_rows)

  • Enables processing of tables larger than available RAM

  • Uses cursor fetching for efficient streaming

pain001.db.validate_db_data module#

Validate SQLite payment data against required-field rules.

pain001.db.validate_db_data.validate_db_data(data: list[dict[str, Any]], message_type: str | None = None) → bool[source]#

Validate the data from a database.

The columns and types come from required_columns(), the same contract the CSV validator enforces.

Parameters:

  • data – The rows to validate, one dictionary per row.

  • message_type – The target message type, e.g. pain.001.001.09, or None when it is not yet known.

Returns:

True if the data is valid, False otherwise.

Return type:

bool