Metadata-Version: 2.4
Name: cb-harun
Version: 0.1.1
Summary: Small reusable building blocks for data engineering: async API extraction, parallel CPU transforms, incremental and full load patterns, memory-safe streaming.
Author-email: Kirandeep Marala <you@example.com>
License: MIT
Project-URL: Homepage, https://github.com/your-username/cb-dataflow
Project-URL: Repository, https://github.com/your-username/cb-dataflow
Project-URL: Issues, https://github.com/your-username/cb-dataflow/issues
Keywords: data-engineering,etl,elt,async,incremental-load,pipeline
Classifier: Development Status :: 4 - Beta
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: MIT License
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Topic :: Database
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Requires-Python: >=3.10
Description-Content-Type: text/markdown
License-File: LICENSE
Requires-Dist: httpx<1.0,>=0.24
Provides-Extra: parquet
Requires-Dist: pyarrow>=12.0; extra == "parquet"
Provides-Extra: pandas
Requires-Dist: pandas>=1.5; extra == "pandas"
Provides-Extra: dev
Requires-Dist: pytest>=7.4; extra == "dev"
Requires-Dist: pytest-asyncio>=0.21; extra == "dev"
Requires-Dist: ruff>=0.1; extra == "dev"
Requires-Dist: build>=1.0; extra == "dev"
Requires-Dist: twine>=4.0; extra == "dev"
Provides-Extra: all
Requires-Dist: cb-dataflow[pandas,parquet]; extra == "all"
Dynamic: license-file

# cb-dataflow

Small, reusable building blocks for data engineering pipelines. Import one piece or all of them.

```bash
pip install cb-dataflow
```

## What is inside

| Module | What it gives you | Core idea |
|---|---|---|
| `cb_dataflow.extract.api` | `AsyncApiExtractor` | Async API calls with retry, backoff, concurrency cap, pagination |
| `cb_dataflow.transform.async_transform` | `amap`, `abatch` | Bounded-concurrency async map that streams results |
| `cb_dataflow.transform.parallel` | `parallel_map`, `parallel_map_batches` | CPU work across processes, without draining the input into RAM |
| `cb_dataflow.patterns.incremental` | `IncrementalLoader` | Watermark / high-water-mark load, commit only after success |
| `cb_dataflow.patterns.full_load` | `FullLoad`, `full_load_file` | Truncate-and-load done safely (staging + atomic swap) |
| `cb_dataflow.patterns.state` | `JsonStateStore`, `SQLiteStateStore` | Where the watermark lives between runs |
| `cb_dataflow.load.writers` | `NdjsonWriter`, `write_parquet_batches` | Streaming writers, one row group per batch |
| `cb_dataflow.memory.chunking` | `batched`, `iter_ndjson`, `downcast_dataframe` | Never hold the whole dataset |
| `cb_dataflow.pipeline` | `run_pipeline` | Wires extract to transform to load in 5 lines |

## Install options

| Command | Gets you |
|---|---|
| `pip install cb-dataflow` | Core only (just `httpx`) |
| `pip install "cb-dataflow[parquet]"` | Adds `pyarrow` for Parquet output |
| `pip install "cb-dataflow[pandas]"` | Adds `pandas` for `downcast_dataframe` |
| `pip install "cb-dataflow[all]"` | Both |
| `pip install "cb-dataflow[dev]"` | pytest, ruff, build, twine |

Heavy libraries sit behind extras so a plain install stays under a second.

## 60-second tour

### Async extraction

```python
import asyncio
from cb_dataflow import AsyncApiExtractor, HttpConfig

async def main():
    cfg = HttpConfig(base_url="https://api.example.com", max_concurrency=8)
    async with AsyncApiExtractor(cfg) as api:
        # one call
        page = await api.get_json("/orders", {"day": "2026-09-08"})

        # many calls, at most 8 in flight, results stream as they land
        reqs = [(f"/customers/{i}", None) for i in range(1000)]
        async for payload in api.fetch_many(reqs):
            ...

        # walk a paged endpoint
        async for items in api.paginate("/posts", items_key="data"):
            ...

asyncio.run(main())
```

### Parallel CPU transform

```python
from cb_dataflow import parallel_map

def parse_row(row):          # must be module level, pools pickle by name
    return {**row, "amount": float(row["amount"])}

for clean in parallel_map(parse_row, raw_rows, chunk_size=500):
    ...
```

### Incremental load

```python
from cb_dataflow import IncrementalLoader, JsonStateStore, write_ndjson

store = JsonStateStore("state/pipeline.json")
inc = IncrementalLoader(store, key="orders", initial="2026-01-01T00:00:00")

rows = fetch_since(inc.start_value())          # ask source only for new rows
new_rows = inc.filter_new(rows, field="updated_at")
write_ndjson(new_rows, "out/orders.ndjson", mode="a")

inc.commit()        # advance the watermark ONLY after the write succeeded
```

Or let the context manager handle commit/rollback:

```python
with IncrementalLoader(store, "orders") as inc:
    write_ndjson(inc.filter_new(rows, field="updated_at"), path, mode="a")
# clean exit -> commit, exception -> rollback
```

### Full load

```python
from cb_dataflow import FullLoad, write_ndjson

with FullLoad("warehouse/customers") as load:
    write_ndjson(all_customers, load.staging_path / "part-0000.ndjson")
# staging swapped in atomically; previous copy kept at warehouse/customers.previous
```

## Incremental vs full load

| | Incremental | Full |
|---|---|---|
| Source needs | A reliable increasing column (`updated_at`, `id`) | Nothing |
| Cost per run | Small, grows with new rows | Grows with total table size |
| Handles hard deletes | No | Yes |
| Failure mode | Re-read the same window (safe) | Old data kept until swap succeeds |
| Delivery guarantee | At-least-once, so the target must dedupe or upsert | Exactly the source snapshot |
| Use when | Big tables, append-heavy sources | Small dimensions, no change column, deletes matter |

## Where the memory savings come from

| Habit | Instead of |
|---|---|
| Generators everywhere | Building a list of all records |
| `batched(...)` before writing | One write call per row, or one giant write |
| `amap(concurrency=N)` | `asyncio.gather(*a_million_tasks)` |
| `parallel_map` submits `workers * 2` chunks | `executor.map` over the whole iterable |
| `iter_ndjson` / `iter_csv_chunks` | `json.load(open(f))` |
| Parquet, one row group per batch | Whole table in RAM before writing |
| `downcast_dataframe` | Default int64/float64/object dtypes |
| `slots=True` dataclasses | Per-instance `__dict__` |

## Run the demo

```bash
pip install -e ".[dev]"
python examples/demo_incremental.py     # first run loads everything
python examples/demo_incremental.py     # second run loads only new hours
```

It pulls real hourly weather for six Indian cities from Open-Meteo (no API key),
flattens it, enriches it across processes, and loads it incrementally.

## Tests

```bash
pytest              # 65 tests, no network needed
```

API tests use `httpx.MockTransport`, so the suite is fast and never flaky.

## License

MIT
