Metadata-Version: 2.4
Name: pyairflowtester
Version: 0.5.1
Summary: Unified Airflow + dbt reliability and quality platform
Project-URL: Homepage, https://github.com/mullassery/pyairflowtester
Project-URL: Repository, https://github.com/mullassery/pyairflowtester
Project-URL: Bug Tracker, https://github.com/mullassery/pyairflowtester/issues
Author: PyAirflowTester Contributors
License-Expression: Apache-2.0
License-File: LICENSE
Keywords: airflow,dbt,quality,reliability,testing
Classifier: Development Status :: 3 - Alpha
Classifier: Intended Audience :: Developers
Classifier: Operating System :: OS Independent
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 :: Software Development :: Quality Assurance
Requires-Python: >=3.10
Requires-Dist: click>=8.1.7
Requires-Dist: rich>=13.7.0
Provides-Extra: all
Requires-Dist: black>=23.12.0; extra == 'all'
Requires-Dist: fastapi>=0.110.0; extra == 'all'
Requires-Dist: httpx>=0.24.0; extra == 'all'
Requires-Dist: jinja2>=3.1.0; extra == 'all'
Requires-Dist: mypy>=1.7.1; extra == 'all'
Requires-Dist: pre-commit>=3.5.0; extra == 'all'
Requires-Dist: pytest-asyncio>=0.21.1; extra == 'all'
Requires-Dist: pytest-cov>=4.1.0; extra == 'all'
Requires-Dist: pytest>=7.4.3; extra == 'all'
Requires-Dist: ruff>=0.1.11; extra == 'all'
Requires-Dist: uvicorn>=0.27.0; extra == 'all'
Provides-Extra: dev
Requires-Dist: black>=23.12.0; extra == 'dev'
Requires-Dist: httpx>=0.24.0; extra == 'dev'
Requires-Dist: mypy>=1.7.1; extra == 'dev'
Requires-Dist: pre-commit>=3.5.0; extra == 'dev'
Requires-Dist: pytest-asyncio>=0.21.1; extra == 'dev'
Requires-Dist: pytest-cov>=4.1.0; extra == 'dev'
Requires-Dist: pytest>=7.4.3; extra == 'dev'
Requires-Dist: ruff>=0.1.11; extra == 'dev'
Provides-Extra: web
Requires-Dist: fastapi>=0.110.0; extra == 'web'
Requires-Dist: jinja2>=3.1.0; extra == 'web'
Requires-Dist: uvicorn>=0.27.0; extra == 'web'
Description-Content-Type: text/markdown

# PyAirflowTester: Airflow & dbt Static Analysis + Dependency Intelligence

## Problem

Airflow DAGs and dbt models accumulate risk silently: a hardcoded secret,
a missing SLA, a circular dependency, an untested high-importance model —
none of it shows up until something breaks in production, and there's
rarely a single tool that checks both the orchestration layer (Airflow)
and the transformation layer (dbt) together.

## Solution

Static analysis and dependency-graph tooling for Airflow DAGs and dbt projects, plus a
library-level dependency intelligence toolkit (impact analysis, blast radius, risk scoring,
observability). Ships as a pure-Python CLI, with an optional web dashboard.

[![PyPI](https://img.shields.io/pypi/v/pyairflowtester)](https://pypi.org/project/pyairflowtester/)
[![CI](https://github.com/Mullassery/PyAirflowTester/actions/workflows/ci.yml/badge.svg)](https://github.com/Mullassery/PyAirflowTester/actions/workflows/ci.yml)
[![License: Apache 2.0](https://img.shields.io/badge/License-Apache%202.0-blue.svg)](./LICENSE)

## Use cases

- **Gating CI on Airflow/dbt risk** — `pyairflowtester scan --format sarif`
  feeds GitHub code scanning directly; `score --compare main` catches a PR
  that raises risk relative to the base branch.
- **Finding blast radius before a deploy** — `dependency blast-radius -n
  <node_id>` answers "is this safe to ship" from a real dependency graph,
  not a guess.
- **A dashboard over DAG/dbt health without standing up a database** —
  `pyairflowtester serve` renders real HTML from the same graph the CLI
  builds, no separate service to run.
- **Not yet a good fit for:** correlating findings against a live Airflow
  metadata DB / dbt run history — the `Analyzer`/`connect` runtime-
  correlation path is an explicit, fail-fast stub today, not a working
  feature (see [What does not work (yet)](#what-does-not-work-yet--please-read-before-relying-on-this)).

## What actually works today

- **`pyairflowtester scan`** — runs 33 static analysis rules against Airflow DAG source files,
  a dbt `manifest.json`, and/or an `airflow.cfg`, and reports violations (security secrets,
  hardcoded connections, missing SLAs, circular dependencies, config misconfigurations, etc).
- **`pyairflowtester rules`** — lists the full rule catalog.
- **`pyairflowtester score`** — aggregate risk score from scan findings.
- **`pyairflowtester dependency ...`** — build a unified dependency graph from DAG files and a
  dbt manifest, then query impact/blast-radius/lineage/cycles/orphans/risk-score over it.
- **`pyairflowtester serve`** — launches a real FastAPI + uvicorn web dashboard (optional
  `pip install pyairflowtester[web]`) that builds the same dependency graph as `dependency
  build` and renders `DashboardBuilder` output as actual browsable HTML pages (a node list at
  `/`, a per-node dashboard at `/nodes/<node_id>`, and an overall `/health` page) — not a JSON
  dump, real HTML tables. See `python/pyairflowtester/web/app.py`.
- The `pyairflowtester.dependency_intelligence` package (usable as a library, see examples
  below) for ownership tracking, schema-evolution tracking, SLA validation, test-coverage
  analysis, anomaly detection, recommendations, and observability primitives
  (metrics/alerts/events/dashboards) — these operate on the graph you build, not on a live
  system. `DashboardBuilder`'s output now also has a real HTML frontend via `serve` above,
  not just plain dicts for programmatic use.

## What does not work (yet) — please read before relying on this

- **Runtime correlation (`Analyzer` class) is not implemented.** It's meant to correlate
  findings against a live Airflow metadata database and dbt run history, but doing that
  honestly requires a real Airflow/dbt instance to build and validate against. Every
  `Analyzer` method (and the `pyairflowtester connect` CLI command) raises
  `AnalyzerNotImplementedError` with a clear message rather than silently returning `[]` and
  pretending nothing was found. This is planned future work, not a working feature.
- **The Rust core (`src/*.rs`, built via PyO3) is not wired into the CLI.** The Python
  package ships as pure Python and does not require Rust or `maturin` to install or run.
  The Rust crate exists as a separate, independent reimplementation of some rule/parsing/
  scoring logic; `python/pyairflowtester/__init__.py` will opportunistically import it if
  you build it yourself (`maturin develop`), but nothing in the CLI path uses it, and it is
  not built or shipped as part of the published package. Treat it as an experiment, not a
  supported acceleration layer.
- **Correction (2026-09-19): this section previously claimed `FailurePredictionEngine` and
  `HealthScoreCalculator` hardcode their test-coverage inputs — that claim is out of date and
  was wrong as of this pass.** Reading `python/pyairflowtester/dependency_intelligence/
  intelligence.py` directly: `predict_node_failure` (line ~133) calls
  `self.test_analyzer.analyze_coverage(node_id).total_tests`, and `_calculate_test_score`
  (line ~437-447) calls the same real `TestCoverageAnalyzer` per node — neither is a hardcoded
  constant. This is also locked in by a regression test,
  `test_test_score_varies_with_real_coverage_data` in
  `python/tests/test_dependency_intelligence_phase2.py`, which asserts the score differs
  between no/partial/full coverage fixtures (verified passing: `pytest
  python/tests/test_dependency_intelligence_phase2.py -k test_score_varies -v`). The one
  real, remaining simplification: `predict_node_failure`'s historical-failure-rate factor
  still assumes a fixed 30-day window (`days_of_data = 30` in `intelligence.py`) rather than
  computing it from actual event timestamps — that part of the original claim was accurate.
  Everything else under "Dependency Intelligence" below (ownership, schema evolution, SLA
  validation, test-coverage analysis via `TestCoverageAnalyzer`, anomaly detection,
  observability) operates on real data you feed it.

## Installation

```bash
pip install pyairflowtester
# or with uv
uv pip install pyairflowtester

# Verify installation
pyairflowtester --version
```

Pure Python — no Rust toolchain required. The CLI's core commands (`scan`, `score`, `rules`,
`dependency ...`) have no dependencies beyond `click`/`rich`. The web dashboard (`serve`) is
optional and pulls in `fastapi`/`uvicorn`/`jinja2`:

```bash
pip install "pyairflowtester[web]"
```

For development:

```bash
git clone https://github.com/mullassery/pyairflowtester.git
cd pyairflowtester
pip install -e ".[dev]"
```

## Static Analysis: `scan`

```bash
# Scan DAGs, a dbt project, and/or airflow.cfg
pyairflowtester scan . --dags dags/ --dbt dbt/ --airflow-cfg airflow.cfg

# Output formats
pyairflowtester scan . --format json --output results.json
pyairflowtester scan . --format html --output report.html
pyairflowtester scan . --format sarif --output results.sarif  # For GitHub code scanning

# Filter results
pyairflowtester scan . --dags dags/ --severity critical
pyairflowtester rules --category security
```

Rule catalog (33 rules, see `pyairflowtester rules` for the live list):

- **AFW001-AFW015** — DAG source-code rules: circular dependencies (real graph-cycle
  detection over parsed `>>`/`<<`/`set_upstream`/`set_downstream` edges, not a regex
  backreference hack), missing SLAs, expensive imports, excessive task counts, risky
  catchup config, default pool usage, hardcoded connection IDs, **hardcoded secrets**,
  excessive retries, sensor timeouts, branch complexity, missing docs, missing alerting,
  deprecated operators.
- **DBT001-DBT003** — dbt manifest rules: missing tests, redundant tests, untested
  high-importance models (derived from the manifest's actual `test.*` nodes and their
  `depends_on`/`attached_node`, not a nonexistent manifest field).
- **CFG001-CFG015** — `airflow.cfg` audit rules: executor choice, pool sizing, concurrency,
  queueing, log retention, encryption, TLS, RBAC, scheduler/worker settings, log storage,
  backups, DAG folder location.

Every rule is evaluated in isolation: if one rule throws, it logs a warning and the rest of
the rules still run and still report their findings for that file.

## Dependency Intelligence

### CLI

```bash
pyairflowtester dependency build --dags dags/ --dbt-manifest dbt/target/manifest.json
pyairflowtester dependency impact <node_id> --depth 10
pyairflowtester dependency lineage --dags dags/
pyairflowtester dependency blast-radius -n <node_id>
pyairflowtester dependency detect-cycles --dags dags/
pyairflowtester dependency detect-orphans --dags dags/
pyairflowtester dependency risk-score --dags dags/ --top 20
```

### As a library

```python
from pyairflowtester.dependency_intelligence import (
    UnifiedGraphBuilder,
    ImpactAnalysisEngine,
    BlastRadiusEngine,
)

# Build a unified graph from DAG files + a dbt manifest
graph = UnifiedGraphBuilder.build_unified_graph(
    dag_files=["dags/my_dag.py"],
    dbt_manifest="dbt/target/manifest.json",
)

# Analyze impact of changing a node
impact = ImpactAnalysisEngine(graph).analyze("dag_my_dag")
print(f"Impact Score: {impact.impact_score:.1%}")
print(f"Impacted Nodes: {len(impact.impacted_nodes)}")

# Calculate deployment risk
blast = BlastRadiusEngine(graph).analyze(["dag_my_dag"])
print(f"Blast Radius: {blast.blast_radius} nodes")
print(f"Safe to Deploy: {'Yes' if blast.deployable else 'No'}")
```

```python
from pyairflowtester.dependency_intelligence import RiskScoringEngine

engine = RiskScoringEngine(graph)
scores = engine.score_all_nodes()

high_risk = sorted(scores.items(), key=lambda x: x[1].risk_score, reverse=True)[:10]
for node_id, score in high_risk:
    print(f"{node_id}: Risk {score.risk_score:.1f}/10")
```

```python
from pyairflowtester.dependency_intelligence import (
    MetricsCollector, AlertManager, EventLogger, DashboardBuilder,
)

# These operate on data you feed them (e.g. from your own Airflow listener/webhook),
# not on a live connection this library establishes itself.
metrics = MetricsCollector()
alerts = AlertManager(graph)
events = EventLogger(graph)

events.log_execution(
    node_id="fact_orders", status="success", duration_ms=1250,
    start_time=..., end_time=...,
)
alerts.set_threshold("fact_orders", "execution_time", warning=5000, critical=10000)

builder = DashboardBuilder(graph, metrics, alerts, events)
dashboard = builder.build_health_dashboard()
```

## Web Dashboard: `serve`

`DashboardBuilder` above returns plain dicts for programmatic use. `pyairflowtester serve`
serves that same output as a real, browsable HTML dashboard — a genuinely minimal app (FastAPI
+ Jinja2-rendered HTML, no JS framework), not a JSON viewer:

```bash
pip install "pyairflowtester[web]"
pyairflowtester serve --dags dags/ --dbt-manifest manifest.json --port 8080
# then open http://127.0.0.1:8080/
```

Routes:

- `GET /` — lists every node in the dependency graph (DAGs, tasks, dbt models, ...), with its
  type, severity, owner, and upstream/downstream counts, linking to its dashboard.
- `GET /nodes/{node_id}` — renders `DashboardBuilder.build_node_dashboard(node_id)` as HTML
  (execution metrics, reliability/failure rate, active alerts, recent events). 404s for an
  unknown node ID.
- `GET /health` — renders `DashboardBuilder.build_health_dashboard()`: graph-wide stats, top
  failing nodes, slowest nodes.

The graph is built once at startup from `--dags`/`--dbt-manifest` (same source-collection logic
as `pyairflowtester dependency build`). Metrics/alerts/events start empty unless you feed them
programmatically (as in the snippet above) before calling `create_app()` yourself — `serve`
itself doesn't fabricate execution history. Implementation: `python/pyairflowtester/web/app.py`;
tests: `python/tests/test_web_app.py` (uses FastAPI's `TestClient`, no real socket bound).

## Architecture

Two real, independent things live in this repo:

1. **The Python CLI (`python/pyairflowtester/`)** — this is what `pip install pyairflowtester`
   ships and what every command above actually runs: `Scanner` (33 rules), `ReportGenerator`,
   `Scorer`, and the `dependency_intelligence` package (graph model, parsers, analytics
   engines, observability primitives). This is the supported, tested path.
2. **A Rust crate (`src/*.rs`)** — a separate, partial reimplementation of some of the same
   rule/parsing/scoring logic using PyO3 bindings (`pyairflowtester._core`). It is **not**
   built or used by the published package or the CLI. `__init__.py` imports it opportunistically
   and falls back to `None` if it isn't present, which is the normal case for anyone who just
   `pip install`s this package. Building the extension yourself (`maturin develop`) does not
   change the CLI's behavior — nothing in the CLI calls into it.

The `Analyzer` class (runtime correlation against live Airflow/dbt) is a stub that raises
`AnalyzerNotImplementedError` — see "What does not work (yet)" above.

## Status

**Proof of concept, actively fixed up.** The static-analysis CLI path (`scan`, `rules`,
`score`, `dependency ...`) works end-to-end and is covered by an automated test suite.
Runtime correlation is explicitly not implemented (fails fast, doesn't fake results). The
Rust core is not part of the supported path.

- Test suite: `python/tests/`, run with `pytest` from the repo root — **198 tests passing,
  1 skipped** with just `pip install -e ".[dev]"` (verify yourself: `pytest python/tests/
  -v`), or **205 passing, 0 skipped** with `pip install -e ".[dev,web]"` (both verified
  2026-09-19 on Python 3.11). The skipped test is in `test_web_app.py`, which is skipped
  automatically if the optional `web` extra isn't installed. Line coverage is ~71-74%
  overall (`--cov-report=term-missing`), but it's uneven: `cli.py` and
  `dependency_intelligence/cli.py` (the actual CLI entry points users run) show **0%**
  coverage — every command is exercised indirectly through the underlying classes, not
  through the CLI wiring itself, so a broken `click` option or argument-passing bug in the
  CLI layer would not be caught by the test suite. `report.py` (20%) and `rules/dbt.py`
  (22%) are also thin. See ROADMAP_HONEST.md for specifics.
- Static rules: 33, all wired into `scan` (previously most of the catalog — the
  `dag_advanced.py` rules including secrets detection, and all of `config.py` — was defined
  but never actually invoked by `scan`).

## CLI Reference

```bash
pyairflowtester scan . --dags dags/ --dbt dbt/ --airflow-cfg airflow.cfg --format html
pyairflowtester score . --compare main
pyairflowtester rules --category reliability --severity critical
pyairflowtester dependency build --dags dags/ --dbt-manifest manifest.json
pyairflowtester dependency impact <node_id> --depth 10
pyairflowtester dependency lineage --format mermaid
pyairflowtester dependency blast-radius -n <node_id>
pyairflowtester dependency detect-cycles
pyairflowtester dependency detect-orphans
pyairflowtester dependency risk-score --top 20
pyairflowtester connect --airflow-home $AIRFLOW_HOME  # currently: reports "not implemented"
pyairflowtester serve --dags dags/ --dbt-manifest manifest.json --port 8080  # requires [web] extra
```

## Requirements

- Python 3.10+
- For Airflow integration: Airflow 2.0+ (only used to shape the DAG source patterns the
  rules look for; Airflow itself is not a runtime dependency)
- For dbt integration: a dbt `manifest.json` (dbt itself is not a runtime dependency)

## Roadmap

Honestly scoped, in priority order:

- Runtime correlation (`Analyzer`): connect to a live Airflow metadata DB and dbt run
  history, replace the current fail-fast stub with real analysis. Needs a live
  Airflow/dbt instance to build and validate against.
- Decide the Rust core's fate: either wire `pyairflowtester._core` into the CLI for real
  (bigger architectural change — would need the two rule/parsing implementations
  reconciled) or drop it to avoid maintaining two parallel implementations.
- Broader dbt manifest coverage, more config-audit rules, richer report formats.
- L2 (Redis) and L4 (DuckDB) cache tiers — `dependency_intelligence/cache.py` now
  has real L1 (in-memory) and L3 (SQLite) tiers with event-driven invalidation (see
  below); Redis/DuckDB would need this otherwise dependency-light package
  (`click`, `rich` only) to take on a heavier/external dependency, so they're left
  for when there's an actual multi-instance-production use case driving it.

## Tiered caching and the dynamic-DAG fallback

`DependencyGraphEngine` accepts an optional `cache=` (a `TieredCache` from
`dependency_intelligence/cache.py`) to persist expensive analyses —
`detect_cycles()`, `get_strongly_connected_components()` — across calls, and
across separate process runs if you back it with a `SqliteCache`:

```python
from pyairflowtester.dependency_intelligence.cache import TieredCache, SqliteCache
from pyairflowtester.dependency_intelligence.graph import DependencyGraphEngine

cache = TieredCache(l3=SqliteCache("~/.cache/pyairflowtester/graph_cache.db"))
engine = DependencyGraphEngine(graph, cache=cache)
cycles = engine.detect_cycles()  # cached by content hash of the graph
```

For DAGs the static AST parser can't see into (built via factory functions,
dynamic loops, or `exec`/`eval`), `dependency_intelligence/runtime_import.py`
adds a sandboxed fallback that actually imports the file in an isolated,
resource-limited subprocess and reads back the real, resolved task graph.
Requires the optional `airflow` package to be installed (this project still
doesn't take it on as a hard runtime dependency):

```python
from pyairflowtester.dependency_intelligence.runtime_import import parse_dag_file_with_fallback

dag_id, task_ids, dependencies = parse_dag_file_with_fallback("dags/dynamic_dag.py")
```

## Contributing

Contributions welcome. Please submit pull requests to GitHub.

## License

This project is licensed under the [Apache License 2.0](LICENSE).

## Contact

For issues, questions, or feature requests: https://github.com/mullassery/pyairflowtester/issues
