asflow 0.0.5
A lightweight, asynchronous workflow runner for ETL pipelines
Requires Python: >=3.11

# AsFlow

**AsFlow** (short for "Async workflow") is a lightweight, asynchronous workflow runner built in pure Python. It's designed for ETL (Extract–Transform–Load) pipelines with a focus on minimal setup, fast iteration, and seamless integration with tools like [Polars](https://pola.rs) and [Streamlit](https://streamlit.io).

![ScreenRecording](https://raw.githubusercontent.com/k24d/asflow/main/docs/assets/img/ScreenRecording.webp)

### Key Features

- 🐍 Pure Python, single-process — no external scheduler or services required
- ⚙️ Asynchronous by design — built on `asyncio` for parallel, non-blocking execution
- 📊 Rich console output — powered by [Rich](https://rich.readthedocs.io) for clean logs and progress bars
- 🔄 Built for data engineering — integrates naturally with [Daft](https://www.getdaft.io), [DuckDB](https://duckdb.org), and [Polars](https://pola.rs)

## Installation

```
% pip install asflow
```

## Quick Start

**AsFlow** lets you build ETL workflows using standard `async` Python functions. Just add the `@flow` and `@flow.task` decorators—everything else behaves like regular Python.

Here’s a minimal example:

```python
import asyncio
import duckdb
from asflow import Flow

flow = Flow(verbose=True)

# Extract: simulate saving raw data
@flow.task(on="words/*.jsonl.gz")
async def extract(word):
    await asyncio.sleep(1)  # Simulate I/O-bound operation
    flow.task.write({"word": word, "count": 1})

# Transform: read raw files into DuckDB
@flow.task
def transform():
    return duckdb.read_json("words/*.jsonl.gz")

# Define the workflow
@flow
async def main():
    words = ["Hello", "World"]

    # Run extractions concurrently
    async with asyncio.TaskGroup() as tg:
        for word in words:
            tg.create_task(extract(word))

    print(transform())

if __name__ == "__main__":
    asyncio.run(main())
```

Running this script produces output like:

```console
% python main.py
[12:34:56] Task extract('Hello') finished in 1.00s
           Task extract('World') finished in 1.01s
           Task transform() finished in 0.00s
┌─────────┬───────┐
│  word   │ count │
│ varchar │ int64 │
├─────────┼───────┤
│ Hello   │     1 │
│ World   │     1 │
└─────────┴───────┘
```

### What’s Happening?

- `extract()` simulates downloading or generating raw data. It’s async, so multiple calls run in parallel.
- `transform()` uses DuckDB to load the saved data into a queryable table.
- Results from `extract()` are written to disk (`words/*.jsonl.gz`), so they’re skipped on future runs if the file already exists—making your workflow faster and more reliable.

### How AsFlow Helps
- ✅ Native support for both **synchronous and asynchronous tasks**
- 🔁 Built-in **retries** for transient failures
- 🧵 Configurable **concurrency limits** to avoid API throttling
- 📦 Persistent storage of **raw data** (e.g., JSON, CSV, text)
- 🗜️ Automatic support for **compressed files** like `.gz` and `.zst`
- 📊 **Rich-powered** logging with progress bars and status indicators

## Documentation

- [Overview](https://asflow.readthedocs.io/en/latest/)
- [Tutorial](https://asflow.readthedocs.io/en/latest/tutorial/)