Metadata-Version: 2.4
Name: apache-airflow-providers-slate
Version: 0.2.0
Summary: Apache Airflow provider for Slate — high-throughput multi-cloud data transfers
License: Apache-2.0
Project-URL: Homepage, https://github.com/tsushanth/slate
Requires-Python: >=3.9
Description-Content-Type: text/markdown
Requires-Dist: apache-airflow>=2.6
Requires-Dist: slate-sdk>=0.2.0

# apache-airflow-providers-slate

Apache Airflow provider for [Slate](https://github.com/tsushanth/slate) — move datasets and model weights between object stores from your Airflow DAGs at 984 MB/s, 4.4× faster than `aws s3 cp`.

## Install

```bash
pip install apache-airflow-providers-slate
```

## Setup

Add a connection in the Airflow UI (or via env var):

```
Connection ID:   slate_default
Connection Type: HTTP
Host:            http://your-slate-api-host
Port:            3030
```

Or via environment variable:
```bash
AIRFLOW_CONN_SLATE_DEFAULT='{"conn_type": "http", "host": "http://localhost", "port": 3030}'
```

Make sure `slate-api` is running:
```bash
DATABASE_URL=sqlite:slate.db?mode=rwc slate-api
```

## Usage

```python
from airflow import DAG
from airflow.utils.dates import days_ago
from apache_airflow_providers_slate.operators.slate import SlateTransferOperator

with DAG(
    dag_id="ml_data_pipeline",
    schedule="@daily",
    start_date=days_ago(1),
    catchup=False,
) as dag:

    # Ingest raw dataset from S3 to GCS staging
    ingest = SlateTransferOperator(
        task_id="ingest_dataset",
        src="s3://raw-data/datasets/imagenet/",
        dst="gs://ml-staging/datasets/imagenet/",
    )

    # Copy model weights to GPU node — runs after ingest
    copy_weights = SlateTransferOperator(
        task_id="copy_weights_to_gpu",
        src="gs://ml-staging/weights/llama-3-70b/",
        dst="/mnt/nvme/weights/llama-3-70b/",
        priority=10,          # pick up before lower-priority jobs
        max_attempts=5,       # retry up to 5 times
        poll_interval=10.0,
    )

    ingest >> copy_weights
```

The operator returns job metadata via XCom so downstream tasks can reference it:

```python
def use_result(**context):
    result = context["ti"].xcom_pull(task_ids="ingest_dataset")
    print(f"Transferred {result['bytes_transferred']} bytes")
    print(f"Peak speed: {result['peak_throughput_mbps']:.1f} MB/s")
```

## Supported stores

| URL scheme | Provider |
|---|---|
| `s3://bucket/prefix` | AWS S3, MinIO, Cloudflare R2 |
| `gs://bucket/prefix` | Google Cloud Storage |
| `az://container/prefix` | Azure Blob Storage |
| `/path` or `file:///path` | Local filesystem |

Any combination of source and destination works.

## Operator reference

```python
SlateTransferOperator(
    task_id="...",
    src="s3://...",             # required
    dst="gs://...",             # required
    slate_conn_id="slate_default",  # Airflow connection ID
    priority=0,                 # higher = picked up sooner
    max_attempts=3,             # retries with exponential backoff
    poll_interval=5.0,          # seconds between status polls
    transfer_timeout=None,      # raise after N seconds (None = no limit)
)
```
