dataloom-engine 0.7.0

DataLoom: a lightweight and efficient thread orchestration engine for data pipelines.

🧵 DataLoom

A lightweight, efficient and safe multi-threaded orchestration engine for Python.

🇧🇷 Versão em português

DataLoom is a library designed to process data streams using the Producer-Consumer pattern with multiple threads (Weavers). It abstracts away the complexity of queues, locks and lifecycle management, letting you focus purely on your data transformation logic.

Why DataLoom?

⚡ Performance: real parallel processing for I/O-bound workloads. 🧩 Simplicity: an intuitive API inspired by the weaving metaphor. 🛡️ Safety: thread-safety guaranteed by design across the whole pipeline. 📦 Lightweight: zero-dependency core, ready to run in any Python environment.

Why not just ThreadPoolExecutor?

Fair question — the stdlib solves the parallelism, but not the pipeline. concurrent.futures gives you a thread pool; everything around it is on you:

You need...	With the stdlib	With DataLoom
A continuous producer-consumer flow	Queue + hand-rolled loops	Loom + Source
Backpressure (producer faster than consumers)	Manual Queue(maxsize=...)	Built-in, configurable
Clean shutdown (drain queue, close resources)	Manual sentinels and joins	stop() / with Loom(...)
Workers that survive errors and report them	try/except in every worker	Centralized hooks.on_error
Per-item processing metrics	Manual instrumentation	hooks.on_batch_processed
Safe concurrent file writes	Manual locking	Ready-made sinks (JSON, CSV, callback)

If your use case is "apply a function to a list and collect the results", use ThreadPoolExecutor.map — it is the right tool. DataLoom is for continuous or long-running flows where lifecycle, resilience and observability matter. And, like every thread-based solution on CPython, the parallelism gains apply to I/O-bound workloads; for CPU-bound work, prefer multiprocessing.

✨ Features

• Cohesive API: aligned concepts (Loom, Weaver, Sink).
• Concurrent: automatically manages multiple Weavers (threads) in parallel.
• Safe: built-in sinks are thread-safe and exceptions are typed (LoomError).
• Resilient: processing errors don't kill the Weavers and are reported through hooks.
• Manageable: Loom is a context manager — with Loom(...) as loom: guarantees automatic stop().
• Backpressure: the task queue is bounded by default (queue_maxsize), preventing unbounded memory growth.
• Observable: lifecycle hooks and per-batch metrics (on_batch_processed with result and duration), plus built-in logging.

📦 Installation

pip install dataloom-engine

The engine core has zero dependencies. The demo RandomNumPySource and StatisticsProcessor (used in the quick start below) need NumPy, which ships as an optional extra:

pip install "dataloom-engine[numpy]"

⚠️ Mind the name: the package installs the dataloom_engine module (import dataloom_engine). Don't confuse it with the dataloom package on PyPI, which is an ORM by another author and unrelated to this project.

To develop or use the latest version from the repository:

git clone https://github.com/dionipadilha/dataloom.git
cd dataloom
pip install -e .

🚀 Quick Start

Here is a complete example of weaving a data pipeline:

from pathlib import Path
import numpy as np
from dataloom_engine import (
    Loom,
    LoomConfig,
    LoomLogs,
    Processor,
    JsonFileSink,
    ThreadedBufferedSink
)
from dataloom_engine.sources import RandomNumPySource

# 1. Define your processing logic (stateless)
class MyFilterProcessor(Processor):
    def process(self, batch: np.ndarray) -> dict:
        # 'batch' is a numpy array sized according to the config
        avg = float(batch.mean())
        return {
            "processed_items": len(batch),
            "average_value": avg,
            "status": "high" if avg > 0.5 else "low"
        }

# 2. Initial setup
if __name__ == "__main__":
    # Configure console logging
    LoomLogs.setup()

    # Engine parameters
    config = LoomConfig(
        output_dir=Path("./data_out"),
        batch_size=100,      # Process 100 items at a time
        interval_seconds=1   # Produce a new batch every second
    )

    # 3. Data source
    # Decoupled from the engine, enabling custom sources (DB, CSV, S3)
    source = RandomNumPySource(config)

    # 4. Buffered file sink
    # ThreadedBufferedSink keeps I/O from blocking the processing
    file_sink = JsonFileSink(config.output_dir)
    sink = ThreadedBufferedSink(file_sink)

    # 5. Initialize the Loom (the orchestrator)
    loom = Loom(
        config=config,
        processor=MyFilterProcessor(),
        sink=sink,
        source=source,
        num_weavers=4  # 4 threads working in parallel
    )

    print("🧵 DataLoom started! Press Ctrl+C to stop.")
    try:
        # The context manager guarantees stop() and resource cleanup,
        # even on exceptions or Ctrl+C
        with loom:
            loom.start()
    except KeyboardInterrupt:
        print("\n🛑 Loom stopped.")

More runnable, real-world examples — API enrichment with parallel speedup, a resilient file-processing pipeline, a buffered sensor stream — live in the examples/ directory.

🏗️ Architecture

DataLoom is built around a weaving metaphor:

• Loom: the main machine. Manages the task queue and the lifecycle.
• Weaver: the worker threads. They take the raw material (a batch), process it and deliver it.
• Processor: the business logic. Turns raw data into information.
• Sink: the final destination. Where the finished product is deposited (e.g. JsonFileSink, CsvFileSink, or any destination via CallbackSink).

🧶 Threading model

Knowing which thread runs each extension point is what makes writing safe Processors, Sinks and Hooks straightforward:

Your code	Runs on
Source.__iter__	the thread that called loom.start() (producer)
Processor.process / Sink.send	the Weaver threads, concurrently
hooks.on_error / hooks.on_batch_processed	the Weaver threads, concurrently
hooks.on_start / hooks.on_stop / Sink.close	the thread that called start() / stop()

Practical consequences:

• Processors and Sinks must be thread-safe if they touch shared state (the built-in sinks are; CallbackSink callables must be).
• Hooks run on the hot path: keep on_batch_processed fast and thread-safe.
• Shutdown semantics: stop() (or leaving the with block) drains the items already queued, then joins the Weavers — pass stop(timeout=...) to bound the wait. A pipeline whose source was exhausted ends as COMPLETED; an interrupted one (external stop(), Ctrl+C) ends as STOPPED; a source error ends as FAILED.

📊 Benchmarks

Reproducible, zero-dependency benchmarks live in benchmarks/throughput.py:

python benchmarks/throughput.py

Reference numbers from a 4-vCPU Linux container (treat them as an order of magnitude — run it on your own hardware):

Scenario	Result
I/O-bound, 200 items × 10ms, 8 weavers	7.8x speedup over sequential
No-op processor (pure engine overhead)	~51,000 items/s (~20µs per item)

The rule of thumb the second number gives you: if your per-item work costs less than ~20µs, a plain loop beats any orchestration — DataLoom pays off when each item does real I/O or meaningful work.

🛠️ Development and Testing

To contribute to the project or run the test suite:

1. Install the development dependencies:

   pip install -e ".[dev]"
   

2. Run the test suite (via pytest):

   pytest
   

3. Run the linter and type checker:

   ruff check .
   mypy
   

📄 License

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

🗒️ Version History

Changes in each release are documented in the CHANGELOG.