ingestflow-sdk 1.0.17

A Python SDK for data ingestion and tracking

ingestflow_SDK

Purpose

The SDK provides a simple Python-based framework for ingesting data from multiple sources into target databases while automatically handling:

1. Duplicate file detection (CSV, JSON)
2. Duplicate record detection (automatically detects duplicate records while inserting into the database and handles reprocessing of files with new records)
3. Incremental loading (loads only new data instead of the entire dataset)
4. Metadata tracking (tracks processed files, records, and execution details)
5. Logging (tracks processed files, inserted records, updated records, duplicate files, duplicate records, and rejected records)

---

Workflow

START | Generate RUN_ID | Read File | Validate Schema | Generate File Hash | Duplicate File Check | +--------------------+ | | | Duplicate | New File | | +---------+----------+ | v Record Validation | v Records To Load? | +----+----+ | | | No | Yes | | v v Update Load Mode Metadata | | +---+---+---+---+ | | | | | | | Full Append Inc Merge | | | | | | | +---+---+---+---+ | | +--------------+ | v Automatic Table Creation | v Insert / Update | v Generate Reject File | v Update Metadata | v Execution Summary | END

Module Description

database/connection.py

Responsible for:

• PostgreSQL database connection
• Connection management
• Connection testing

---

readers

csv_reader.py

Responsible for:

• Reading CSV files
• Converting data into Pandas DataFrame

json_reader.py

Responsible for:

• Reading JSON files
• Converting data into Pandas DataFrame

---

trackers

file_tracker.py

Responsible for:

• Generating SHA256 file hash
• Detecting duplicate files
• Tracking processed files

record_tracker.py

Responsible for:

• Incremental load processing
• Merge load processing
• Duplicate record identification
• Update record identification

metadata_manager.py

Responsible for:

• Creating metadata tables
• Tracking SDK executions
• Tracking file processing details
• Tracking load statistics

---

validators

schema_validator.py

Responsible for:

• Validating incoming file structure
• Comparing source schema with target table schema
• Detecting missing or extra columns

data_validator.py

Responsible for:

• Data quality validation
• Empty row detection
• Reject file generation

---

writers

postgres_writer.py

Responsible for:

• Automatic table creation
• Data insertion
• Data update
• Table truncation

---

logging

logger.py

Responsible for:

• Printing execution summary
• Displaying processing statistics
• Displaying execution status

---

utils

run_id_generator.py

Responsible for:

• Generating unique execution IDs

Example:

RUN_20260603_001
RUN_20260603_002
RUN_20260603_003

---

Supported File Types

Current supported source formats:

CSV
JSON

---

Supported Load Types

Full Load

load_type="full"

Behavior:

Truncate target table
Load complete dataset

---

Append Load

load_type="append"

Behavior:

Insert all records
No update operation

---

Incremental Load

load_type="incremental"

Required:

incremental_column="updated_date"

Behavior:

Load only records newer than the maximum value present in the target table

---

Merge Load

load_type="merge"

Required:

primary_keys=["id"]

Behavior:

Insert new records
Update existing records
Ignore unchanged records

---

Metadata Tables

The SDK automatically creates the following metadata tables.

run_metadata

Tracks SDK executions.

Columns:

run_id
start_time
end_time
status
source_file
target_table

---

file_metadata

Tracks processed files.

Columns:

file_id
run_id
file_name
file_hash
record_count
processed_timestamp

---

load_metadata

Tracks load statistics.

Columns:

load_id
run_id
inserted_count
updated_count
duplicate_count
rejected_count

---

Reject File Handling

Invalid records are written to the rejects folder.

Example:

rejects/

RUN_20260603_001_rejects.csv

Sample:

id,name,reject_reason
2,Alice,Invalid Record