pain001.csv package#

CSV operations module for pain001.

Submodules#

pain001.csv.load_csv_data module#

Load payment data from CSV files.

pain001.csv.load_csv_data.load_csv_data(file_path: str, sanitize_formula: bool = False) → list[dict[str, Any]][source]#

Load CSV data from a file.

Parameters:

  • file_path – The path to the CSV file.

  • sanitize_formula – If True, neutralize spreadsheet formula injection characters (‘=’, ‘+’, ‘-’, ‘@’, ‘t’, ‘r’, ‘n’) by prefixing cells with a single quote. Defaults to False to prevent corruption of legitimate financial identifiers and contact strings.

Returns:

A list of dictionaries containing the CSV data.

Return type:

list

Raises:

  • FileNotFoundError – If the file does not exist.

  • OSError – If there is an issue reading the file.

  • UnicodeDecodeError – If there is an issue decoding the file’s content.

  • DataSourceError – If the CSV file is empty.

  • Exception – If path validation fails, the underlying error is logged and re-raised unchanged.

Note

For large files, consider using load_csv_data_streaming() to reduce memory footprint.

pain001.csv.load_csv_data.load_csv_data_streaming(file_path: str, chunk_size: int = 1000, sanitize_formula: bool = False) → Generator[list[dict[str, Any]], None, None][source]#

Load CSV data from a file in chunks for memory-efficient processing.

This function yields chunks of CSV data instead of loading the entire file into memory, making it suitable for large files.

Parameters:

  • file_path – The path to the CSV file.

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

  • sanitize_formula – If True, neutralize spreadsheet formula injection characters (‘=’, ‘+’, ‘-’, ‘@’, ‘t’, ‘r’, ‘n’) by prefixing cells with a single quote. Defaults to False to prevent corruption of legitimate financial identifiers and contact strings.

Yields:

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

Raises:

  • FileNotFoundError – If the file does not exist.

  • OSError – If there is an issue reading the file.

  • UnicodeDecodeError – If there is an issue decoding the file’s content.

  • DataSourceError – If the CSV file is empty.

  • Exception – If path validation fails, the underlying error is logged and re-raised unchanged.

Example

>>> for chunk in load_csv_data_streaming('large_file.csv', chunk_size=500):
...     # Process chunk
...     process_payment_batch(chunk)
Performance:

  • Memory usage: ~90% reduction for large files (10K+ rows)

  • Enables processing of files larger than available RAM

  • Slightly slower than load_csv_data() due to yielding overhead

pain001.csv.validate_csv_data module#

Validate CSV payment data against required-field rules.

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

Validate payment rows against the required-column contract.

The columns and types come from required_columns(): the bundled JSON schema of message_type when given, otherwise every column all bundled schemas describe.

Parameters:

  • data – A list of dictionaries containing the CSV data.

  • 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