pain001.json package#

JSON data loaders for pain001.

Submodules#

pain001.json.load_json_data module#

JSON data loader for payment data.

pain001.json.load_json_data.load_json_data(file_path: str) → list[dict[str, Any]][source]#

Load payment data from JSON file.

Supports both single object and array of objects format.

Parameters:

file_path – Path to JSON file containing payment data.

Returns:

List of payment data dictionaries.

Raises:

  • FileNotFoundError – If file doesn’t exist.

  • DataSourceError – If JSON is malformed, empty, or not an object/array.

Examples

# Array format (preferred) >>> data = load_json_data(‘payments.json’) # [{‘id’: ‘MSG001’, ‘amount’: ‘1000.00’, …}, …]

# Single object format >>> data = load_json_data(‘payment.json’) # Automatically wrapped: [{‘id’: ‘MSG001’, …}]

pain001.json.load_json_data.load_json_data_streaming(file_path: str, chunk_size: int = 1000) → Generator[list[dict[str, Any]], None, None][source]#

Load JSON data in chunks for memory efficiency.

Note: This loads the entire JSON file into memory first, then yields chunks. JSON doesn’t support true streaming due to its structure. For large JSON files, consider converting to JSONL (JSON Lines) format or use Parquet.

Parameters:

  • file_path – Path to JSON file.

  • chunk_size – Number of records per chunk (default: 1000).

Errors from load_json_data (FileNotFoundError, DataSourceError) propagate unchanged.

Yields:

list[dict[str, Any]] – Chunks of payment data dictionaries.

Examples

>>> for chunk in load_json_data_streaming('payments.json', chunk_size=500):
...     process_batch(chunk)
pain001.json.load_json_data.load_jsonl_data(file_path: str) → list[dict[str, Any]][source]#

Load payment data from JSON Lines (.jsonl) file.

JSON Lines format has one JSON object per line, enabling true streaming for large datasets.

Parameters:

file_path – Path to JSONL file.

Returns:

List of payment data dictionaries.

Raises:

  • FileNotFoundError – If file doesn’t exist.

  • DataSourceError – If the JSONL is invalid or empty.

  • Exception – A DataSourceError raised inside the read loop is re-raised unchanged via a bare raise.

Examples

>>> data = load_jsonl_data('payments.jsonl')
pain001.json.load_json_data.load_jsonl_data_streaming(file_path: str, chunk_size: int = 1000) → Generator[list[dict[str, Any]], None, None][source]#

Load JSONL data in true streaming fashion.

This is the preferred method for large JSON datasets as it doesn’t load the entire file into memory.

Parameters:

  • file_path – Path to JSONL file.

  • chunk_size – Number of records per chunk (default: 1000).

Yields:

list[dict[str, Any]] – Chunks of payment data dictionaries.

Raises:

  • FileNotFoundError – If file doesn’t exist.

  • DataSourceError – If the JSONL is invalid.

  • Exception – A DataSourceError raised inside the read loop is re-raised unchanged via a bare raise.

Examples

>>> for chunk in load_jsonl_data_streaming('large_payments.jsonl'):
...     process_batch(chunk)