AtomIngest

A Metadata-Driven Data Ingestion Framework for AtomSQL

AtomIngest is a configurable, zero-boilerplate data ingestion framework designed for fintech environments where data reliability, compliance, and scalability are paramount. Built on the metaprogramming foundations of AtomSQL, it dynamically generates ORM models, validators, and transformation logic entirely from YAML configurations.

---

🚀 Key Features

AtomIngest transforms raw data (CSV, JSON, APIs) into structured, compliant records in AtomSQL-managed databases without writing repetitive model code.

• 📄 YAML-Driven Configuration: Define table schemas, constraints, and pipelines declaratively.
• 🔮 Dynamic Model Generation: Uses Python metaprogramming to create AtomSQL models (ModelMeta) at runtime.
• 🛡️ Fintech-Grade Security:
  • PII Vault: Transparent field-level encryption for sensitive data.
  • Immutable Ledgers: WORM (Write Once, Read Many) support for audit trails.
  • Crypto-Shredding: GDPR compliance via key destruction.
• ✅ Robust Validation: Inject regex, range checks, and custom business rules directly into Field definitions.
• 🔄 Reliability: Built-in support for Idempotency (Upserts), Dead Letter Queues (DLQ), and Transaction Management.
• ⚡ High Performance: Async batching and smart buffering for high-throughput feeds.

---

📦 Installation

pip install atomingest

Requires Python 3.12+ and AtomSQL.

---

⚡ Quick Start

1. Define your Pipeline (trades.yaml)

Create a YAML configuration defining your target table, schema, and validation rules. AtomIngest maps these types directly to atomsql.orm.fields.

target_table: trades
strategy: upsert
business_keys: [trade_id]

schema:
  trade_id:
    type: StringField
    unique: true
    nullable: false
  symbol:
    type: CharField
    max_length: 10
    validation:
      regex: "^[A-Z]{3,5}$"  # Validates ticker symbols (e.g., AAPL)
  amount:
    type: DecimalField
    validation:
      min: 0.01
  executed_at:
    type: DateTimeField

hooks:
  pre_save: "utils.enrich_metadata"

2. Run the Ingestion

Use the CLI or Python API to ingest data. The framework dynamically builds the Trade model and inserts data into your database.

Using CLI:

atomingest run trades.yaml --source data/daily_trades.csv

Using Python:

from atomingest.core import Ingester
from atomsql import Database

# Connect to DB
db = Database("sqlite:///finance.db")

# Initialize Ingester
ingest = Ingester(db)

# Run Pipeline
stats = ingest.run(
    config="trades.yaml",
    source="data/daily_trades.csv"
)

print(f"Ingested: {stats.processed}, Failed: {stats.failed}")

3. Normalization Pipeline

AtomIngest includes a powerful normalization pipeline to clean and standardize raw data before ingestion:

from atomingest.normalization import normalize_csv_file, NormalizationStepConfig

# Define normalization steps
steps = [
    NormalizationStepConfig(step="RemoveBOMStep", config={}, enabled=True),
    NormalizationStepConfig(step="TrimWhitespaceStep", config={"trim_quotes": True}, enabled=True),
    NormalizationStepConfig(step="ColumnNameNormalizationStep", config={"case": "lower"}, enabled=True),
]

# Normalize data
rows, report = normalize_csv_file("messy_data.csv", steps=steps)
print(f"Cleaned {report.rows_emitted} rows with {len(report.warnings)} warnings")

YAML Integration:

target_table: customers

# Embedded normalization configuration
normalization:
  strict: false
  steps:
    - step: TrimWhitespaceStep
      config: {trim_quotes: true, collapse_spaces: true}
      enabled: true
    - step: NullTokenNormalizationStep
      config: {null_tokens: ["", "NA", "NULL", "-"]}
      enabled: true
    - step: BasicTypeCleanupStep
      config:
        numeric_columns: ["age", "amount"]
        date_columns: ["created_date"]
        date_formats: ["%Y-%m-%d", "%m/%d/%Y"]
      enabled: true

schema:
  name: {type: CharField, max_length: 100}
  age: {type: IntegerField}
  amount: {type: DecimalField}
  created_date: {type: DateField}

---

🏗️ Architecture

AtomIngest leverages the dynamic nature of AtomSQL:

1. Schema Loader: Parses YAML and resolves types to AtomSQL classes (e.g., StringField, IntegerField).

2. Metaclass Factory: Uses type() to construct a new class inheriting from atomsql.orm.models.Model, automatically registering it with the Database.

3. Pipeline Runner: Streams data, applies "Validator Injection" decorators, executes hooks, and manages transactions.

4. Reliability Layer: Catches errors per row, routing failed records to a DLQ table while committing valid rows in batches.

---

🗺️ Roadmap

We are currently building out the core phases:

• Phase 1: Core Configuration - YAML Parser & Dynamic Model Factory.

• Phase 2: Validation - Regex/Range constraints & Hook system.

• Phase 3: Reliability - DLQ, Upsert strategies, and Async Batching.

• Phase 4: Security (In Progress) - PII Encryption & Immutable Mixins.

• Phase 5: Orchestration - Topological dependency resolution.

See the full Project Backlog for details.

---

🤝 Contributing

We welcome contributions! Please see CONTRIBUTING.md for details on how to set up the dev environment.

1. Clone the repo.
2. Install dependencies: uv pip install -r pyproject.toml.
3. Run tests: pytest tests/.

---

📄 License

This project is licensed under the MIT License - see the LICENSE file for details.

---

Built with ❤️ for the Fintech Open Source Community.