Metadata-Version: 2.4
Name: dagster-floe
Version: 0.1.9
Summary: Dagster connector for Floe CLI (config-driven ingestion)
Author: Floe contributors
License-Expression: MIT
Project-URL: Homepage, https://github.com/malon64/floe
Project-URL: Repository, https://github.com/malon64/floe
Project-URL: Issues, https://github.com/malon64/floe/issues
Project-URL: Documentation, https://github.com/malon64/floe/tree/main/orchestrators/dagster-floe
Keywords: dagster,orchestration,data-quality,ingestion
Classifier: Development Status :: 5 - Production/Stable
Classifier: Intended Audience :: Developers
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3 :: Only
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Topic :: Software Development :: Libraries
Requires-Python: >=3.10
Description-Content-Type: text/markdown
Requires-Dist: dagster<2,>=1.7
Requires-Dist: jsonschema>=4
Provides-Extra: kubernetes
Requires-Dist: kubernetes>=28; extra == "kubernetes"
Provides-Extra: remote
Requires-Dist: fsspec; extra == "remote"
Requires-Dist: s3fs; extra == "remote"
Requires-Dist: gcsfs; extra == "remote"
Requires-Dist: adlfs; extra == "remote"
Provides-Extra: dev
Requires-Dist: pytest>=7; extra == "dev"

# Floe + Dagster

This folder contains the Dagster connector for Floe.

For local setup of both Dagster and Airflow with isolated virtual environments, see:
- `orchestrators/LOCAL_DEV.md`

Current model:
- Parse-time reads Floe manifests (`floe.manifest.v1`), not Floe YAML.
- Supports single manifest and multi-manifest directory loading.
- Builds one asset per entity and one job per manifest.
- Runs Floe with manifest execution contract and JSON logs.
- Publishes run/entity metadata from NDJSON + summary.

## What Is Implemented

- Manifest-first orchestration (`floe.manifest.v1`).
- Strict schema validation at manifest load time.
- Asset generation from `entities[]` (`asset_key`, `group_name` respected).
- One Dagster job per manifest (stable job names).
- Multi-manifest loading (`*.manifest.json`) with collision checks.
- Local runner (`local_process`) support.
- `execution.defaults.env` and `execution.defaults.workdir` support.
- Floe quality outcomes exposed as Dagster native Asset Checks (`cast_error`, `not_null`, `unique`, `schema_mismatch`, `file_status`) from Floe reports.

## Install

Prereqs:
- Floe installed (either the `floe` CLI binary **or** Docker with a Floe image).
- Python 3.10+.

```bash
pip install dagster-floe
```

## Development install (from this repo)

```bash
python3 -m venv orchestrators/dagster-floe/.venv
source orchestrators/dagster-floe/.venv/bin/activate
pip install -e orchestrators/dagster-floe[dev]
```

## Generate a manifest

```bash
floe manifest generate \
  -c orchestrators/dagster-floe/example/config.yml \
  --output orchestrators/dagster-floe/example/manifest.dagster.json
```

## Run the example (repo-only)

```bash
cd orchestrators/dagster-floe
FLOE_MANIFEST_DIR=./example/manifests dagster dev
```

The example workspace loads `example/definitions.py`, which wires local example files/manifest to the reusable connector APIs.
The repository example includes two manifests by domain:
- `example/manifests/hr.manifest.json`
- `example/manifests/sales.manifest.json`

## Asset checks

Every accepted-output asset automatically gets quality checks registered in Dagster — no extra config needed. After each run, the connector reads the Floe run report and publishes pass/fail results for each entity.

| Check | Fails when |
| --- | --- |
| `file_status` | One or more input files had a failed or error-level processing status (e.g., unreadable file, parse failure) |
| `cast_error` | Type-casting failures exceeded the entity's policy threshold |
| `not_null` | Null values found in non-nullable columns |
| `unique` | Duplicate values found in uniqueness-constrained columns |
| `schema_mismatch` | Incoming file schema is incompatible with the declared schema |

Checks appear in the Dagster UI under each asset, linked to the run that produced them. A `WARN`-severity violation marks the check as a warning; `REJECT` or `ABORT` marks it as a failure.

## Lineage integration

The connector automatically injects two environment variables into every floe subprocess:

- `DAGSTER_RUN_ID` — the current Dagster run ID
- `DAGSTER_JOB_NAME` — the Dagster job name

When lineage is enabled and both variables are present, floe attaches a **parent facet** to its OpenLineage run event. This facet declares that the Floe run is a child of the Dagster run, making the lineage graph show the full Dagster → Floe pipeline hierarchy.

The Floe run event's own `job.name` is derived from `lineage.job_name` in the embedded manifest lineage config, or from the config file stem. To control it explicitly, add `job_name` to your lineage block when generating the manifest:

```yaml
# prod.yml (profile)
lineage:
  url: "http://marquez:5000"
  namespace: "my-data-platform"
  job_name: "orders-pipeline"   # sets job.name on the Floe OpenLineage run event
```

```bash
floe manifest generate -c orders.yml -p prod.yml --output manifests/orders.json
```

See [docs/lineage.md](../../docs/lineage.md) for the full lineage config reference.

## Notes

- This connector does **not** parse YAML directly; it consumes `floe.manifest.v1`.
- Connector logic lives under `src/floe_dagster/`; local wiring for demo lives in `example/definitions.py`.
- For local development without an installed `floe` binary, you can point `LocalRunner` to a custom command, e.g.:
  - `LocalRunner(\"cargo run -p floe-cli --\")`
- Manifest runner support in connector is currently `local_process` only.
- For local setup commands, use `orchestrators/LOCAL_DEV.md`.
- Design notes and future work: `orchestrators/dagster-floe/INTEGRATION_SPEC.md`

## What Is Not Implemented Yet

- Kubernetes/ECS runner adapters.
- Cloud summary loading (`s3://`, `gs://`, `abfs://`).
- Single-process multi-entity fan-out execution mode.

## Releasing

This repo is a monorepo. Floe and this connector are versioned and tagged independently:

- Floe CLI release tags: `vX.Y.Z`
- Dagster connector release tags: `dagster-floe-vX.Y.Z` (triggers the PyPI publish workflow)

Example:

```bash
git checkout main
git pull
git tag dagster-floe-v0.1.0
git push origin dagster-floe-v0.1.0
```
