zoopipe 2026.1.16

ZooPipe is a data processing framework that allows you to process data in a declarative way.

ZooPipe

ZooPipe is a lean, ultra-high-performance data processing engine for Python. It leverages a 100% Rust core to handle I/O and orchestration, while keeping the flexibility of Python for schema validation (via Pydantic) and custom data enrichment (via Hooks).

---

✨ Key Features

• 🚀 100% Native Rust Engine: The core execution loop, including CSV and JSON parsing/writing, is implemented in Rust for maximum throughput.
• 🔍 Declarative Validation: Use Pydantic models to define and validate your data structures naturally.
• 🪝 Python Hooks: Transform and enrich data at any stage using standard Python functions or classes.
• 🚨 Automated Error Routing: Native support for routing failed records to a dedicated error output.
• 📊 Multiple Format Support: Optimized readers/writers for CSV, JSONL, and SQL databases (via SQLx with batch inserts).
• 🔧 Pluggable Executors: Choose between single-threaded or multi-threaded execution strategies.

---

🚀 Quick Start

Installation

uv build
uv run maturin develop --release

Simple Example

from pydantic import BaseModel, ConfigDict
from zoopipe import CSVInputAdapter, CSVOutputAdapter, Pipe


class UserSchema(BaseModel):
    model_config = ConfigDict(extra="ignore")
    user_id: str
    username: str
    email: str


pipe = Pipe(
    input_adapter=CSVInputAdapter("users.csv"),
    output_adapter=CSVOutputAdapter("processed_users.csv"),
    error_output_adapter=CSVOutputAdapter("errors.csv"),
    schema_model=UserSchema,
)

pipe.start()
pipe.wait()

print(f"Finished! Processed {pipe.report.total_processed} items.")

---

📚 Documentation

Core Concepts

• Executors Guide - Choose and configure execution strategies
• Hooks Guide - Transform and enrich data using Python hooks

Hooks

Hooks are Python classes that allow you to intercept, transform, and enrich data at different stages of the pipeline.

📘 Read the full Hooks Guide to learn about lifecycle methods (setup, execute, teardown), state management, and advanced patterns like cursor pagination.

Quick Example

from zoopipe import BaseHook

class MyHook(BaseHook):
    def execute(self, entries, store):
        for entry in entries:
            entry["raw_data"]["checked"] = True
        return entries

[!IMPORTANT] If you are using a schema_model, the pipeline will output the contents of validated_data for successful records.

• To modify data before validation, use pre_validation_hooks and modify entry["raw_data"].
• To modify data after validation (and ensure it reaches the output), use post_validation_hooks and modify entry["validated_data"].

Input/Output Adapters

File Formats

• CSV Adapters - High-performance CSV reading and writing
• JSON Adapters - JSONL and JSON array format support
• Excel Adapters - Read and write Excel (.xlsx) files
• Parquet Adapters - Columnar storage for analytics and data lakes
• Arrow Adapters - Apache Arrow IPC format for high-throughput interoperability

Databases

• SQL Adapters - Read from and write to SQL databases with batch optimization
• SQL Pagination - High-performance cursor-style pagination for large tables
• DuckDB Adapters - Analytical database for OLAP workloads

Messaging Systems

• Kafka Adapters - High-throughput messaging

Advanced

• Python Generator Adapters - In-memory streaming and testing
• Cloud Storage (S3) - Read and write data from Amazon S3 and compatible services
• PipeManager - Run multiple pipes in parallel for distributed processing

---

🛠 Architecture

ZooPipe is designed as a thin Python wrapper around a powerful Rust core:

1. Python Layer: Configuration, Pydantic models, and custom Hooks.
2. Rust Core:
   • Adapters: High-speed CSV/JSON/SQL Readers and Writers with optimized batch operations.
   • NativePipe: Orchestrates the loop, fetching chunks, calling a consolidated Python batch processor, and routing result batches.
   • Executors: Single-threaded or multi-threaded batch processing strategies.

---

📄 License

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