Metadata-Version: 2.5
Name: pyairflowtester
Version: 0.5.0
Summary: Unified Airflow + dbt reliability and quality platform
Project-URL: Homepage, https://github.com/mullassery/pyairflowtester
Project-URL: Documentation, https://pyairflowtester.readthedocs.io
Project-URL: Repository, https://github.com/mullassery/pyairflowtester
Project-URL: Bug Tracker, https://github.com/mullassery/pyairflowtester/issues
Author: PyAirflowTester Contributors
License: Proprietary License — Free to use with explicit attribution
License-File: LICENSE
Keywords: airflow,dbt,quality,reliability,testing
Classifier: Development Status :: 3 - Alpha
Classifier: Intended Audience :: Developers
Classifier: License :: Other/Proprietary License
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

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.

## 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.
- **The `dependency_intelligence` "Phase 3" intelligence engines (`FailurePredictionEngine`,
  `HealthScoreCalculator`) don't actually consult real test-coverage data, despite the
  comments suggesting they should.** `FailurePredictionEngine.predict_node_failure` hardcodes
  `test_count = 0` for every node (so "no test coverage" always contributes to the failure
  score, regardless of what you've fed `TestCoverageAnalyzer`) and assumes a fixed 30-day
  failure window. `HealthScoreCalculator._calculate_test_score` always returns the same fixed
  value (10.0) regardless of the graph. They're importable and exported, but their
  test-coverage inputs are not wired to the real `TestCoverageAnalyzer` yet — treat their
  output as illustrative, not measured. Everything else under "Dependency Intelligence" below
  (ownership, schema evolution, SLA validation, test-coverage analysis via
  `TestCoverageAnalyzer`, anomaly detection, observability) does operate 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 — **205 tests, 0
  failing** (verify yourself: `pytest python/tests/ -v`). The web dashboard tests
  (`test_web_app.py`) are skipped automatically if the optional `web` extra isn't installed.
- 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

Proprietary. See LICENSE file for details.

## Contact

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