Feat/configurable analysis policy

.github/workflows/ci.yaml

@@ -28,5 +28,20 @@ jobs:
       - name: Run Ruff
         run: ruff check .
 
+      - name: Check formatting
+        run: ruff format --check .
+
+      - name: Run Pyright
+        run: pyright
+
       - name: Run unit tests
-        run: python -m pytest -m "not integration" -v
+        run: python -m pytest -m "not integration" -v
+
+      - name: Build package
+        run: python -m build
+
+      - name: Smoke test wheel
+        run: |
+          uv venv /tmp/flowsense-wheel-test
+          uv pip install --python /tmp/flowsense-wheel-test/bin/python dist/*.whl
+          /tmp/flowsense-wheel-test/bin/flowsense --help

README.md

@@ -135,6 +135,48 @@ Then run:
 flowsense analyze <dag_id>
 ```
 
+## Library API
+
+FlowSense can also be used as a Python library through its supported top-level
+API:
+
+```python
+from flowsense import AirflowClient, analyze_dag
+
+with AirflowClient() as source:
+    analysis = analyze_dag(
+        dag_id="flowsense_demo",
+        source=source,
+    )
+
+print(analysis.overall_severity)
+print(analysis.primary_origin)
+```
+
+Analysis behavior can be customized with an immutable policy:
+
+```python
+from flowsense import AnalysisPolicy, MappedTaskAggregation
+
+policy = AnalysisPolicy(
+    minimum_history=10,
+    baseline_window=30,
+    medium_threshold=2.5,
+    high_threshold=4.0,
+    critical_threshold=6.0,
+    mapped_task_aggregation=MappedTaskAggregation.MAX,
+)
+```
+
+`baseline_window` limits the number of historical values used before the current
+run. Mapped task durations can be aggregated with `MAX`, `MEAN`, or `SUM`. The
+same policy options are available through the CLI and MCP tool.
+
+Custom data sources can implement the `DAGDataSource` protocol and be passed to
+`analyze_dag`. Names exported directly from `flowsense` form the supported public
+API. Imports from internal packages such as `flowsense.engine` should be treated
+as implementation details and may change before version 1.0.
+
 ## MCP Server
 
 Start the FlowSense MCP server over stdio:
@@ -183,17 +225,21 @@ ruff format .
 
 ```text
 src/flowsense/
-├── cli/
-├── collector/
+├── application/
+├── domain/
 ├── engine/
 │   ├── drift.py
 │   ├── history.py
 │   ├── impact.py
 │   ├── propagation.py
 │   ├── root_cause.py
 │   └── timing.py
+├── infrastructure/
+│   └── airflow/
+├── cli/
 ├── mcp/
-└── models/
+├── collector/  # backward-compatible imports
+└── models/     # backward-compatible imports
 ```
 
 ## Detection Approach

pyproject.toml

@@ -8,14 +8,14 @@ dependencies = [
     "httpx>=0.27",
     "pydantic>=2.8",
     "numpy>=2.0",
-    "pandas>=2.2",
-    "scipy>=1.14",
     "typer>=0.12",
     "rich>=13.7",
 ]
 
 [project.optional-dependencies]
 dev = [
+    "build>=1.2",
+    "pyright>=1.1.400",
     "pytest>=8.0",
     "ruff>=0.6",
 ]
@@ -38,8 +38,16 @@ package-dir = {"" = "src"}
 [tool.setuptools.packages.find]
 where = ["src"]
 
+[tool.setuptools.package-data]
+flowsense = ["py.typed"]
+
+[tool.pyright]
+include = ["src"]
+typeCheckingMode = "standard"
+pythonVersion = "3.12"
+
 [tool.pytest.ini_options]
 pythonpath = ["src"]
 markers = [
     "integration: tests that require a running Airflow instance",
-]
+]

src/flowsense/__init__.py

@@ -0,0 +1,49 @@
+from importlib.metadata import PackageNotFoundError, version
+
+from flowsense.application import DAGDataSource, analyze_dag
+from flowsense.domain import (
+    DEFAULT_ANALYSIS_POLICY,
+    AnalysisDiagnostic,
+    AnalysisPolicy,
+    DAGAnalysis,
+    DriftResult,
+    FlowSenseError,
+    ImpactClassification,
+    InsufficientHistoryError,
+    InvalidTaskTimingError,
+    MappedTaskAggregation,
+    PropagationResult,
+    RootCauseResult,
+    Severity,
+    TaskImpact,
+    TaskRun,
+)
+from flowsense.infrastructure.airflow import AirflowApiError, AirflowClient
+
+try:
+    __version__ = version("flowsense")
+except PackageNotFoundError:
+    __version__ = "0.0.0"
+
+__all__ = [
+    "DEFAULT_ANALYSIS_POLICY",
+    "AirflowApiError",
+    "AirflowClient",
+    "AnalysisDiagnostic",
+    "AnalysisPolicy",
+    "DAGAnalysis",
+    "DAGDataSource",
+    "DriftResult",
+    "FlowSenseError",
+    "ImpactClassification",
+    "InsufficientHistoryError",
+    "InvalidTaskTimingError",
+    "MappedTaskAggregation",
+    "PropagationResult",
+    "RootCauseResult",
+    "Severity",
+    "TaskImpact",
+    "TaskRun",
+    "__version__",
+    "analyze_dag",
+]

src/flowsense/application/__init__.py

@@ -0,0 +1,4 @@
+from flowsense.application.analyzer import analyze_dag
+from flowsense.application.ports import DAGDataSource
+
+__all__ = ["DAGDataSource", "analyze_dag"]

src/flowsense/application/analyzer.py

@@ -0,0 +1,146 @@
+from __future__ import annotations
+
+from flowsense.application.ports import DAGDataSource
+from flowsense.domain import (
+    DEFAULT_ANALYSIS_POLICY,
+    AnalysisDiagnostic,
+    AnalysisPolicy,
+    DAGAnalysis,
+    InsufficientHistoryError,
+    Severity,
+)
+from flowsense.domain.enums import SEVERITY_SCORE
+from flowsense.engine.drift import calculate_drift
+from flowsense.engine.history import build_duration_history
+from flowsense.engine.impact import classify_task_impact
+from flowsense.engine.propagation import analyze_propagation
+from flowsense.engine.root_cause import select_primary_origin
+from flowsense.engine.timing import (
+    build_handoff_history_with_diagnostics,
+    calculate_handoff_drift,
+)
+
+
+def analyze_dag(
+    dag_id: str,
+    source: DAGDataSource,
+    policy: AnalysisPolicy = DEFAULT_ANALYSIS_POLICY,
+) -> DAGAnalysis:
+    task_runs = source.collect_task_runs(dag_id)
+    duration_history = build_duration_history(
+        task_runs,
+        aggregation=policy.mapped_task_aggregation,
+    )
+    diagnostics: list[AnalysisDiagnostic] = []
+
+    drift_results = {}
+
+    for task_id, durations in duration_history.items():
+        try:
+            drift_results[task_id] = calculate_drift(
+                task_id=task_id,
+                durations=durations,
+                policy=policy,
+            )
+        except InsufficientHistoryError as exc:
+            diagnostics.append(
+                AnalysisDiagnostic(
+                    code="INSUFFICIENT_TASK_HISTORY",
+                    subject_id=task_id,
+                    message=str(exc),
+                )
+            )
+
+    dependencies = source.get_dag_dependencies(dag_id)
+
+    handoff_history_result = build_handoff_history_with_diagnostics(
+        task_runs=task_runs,
+        dependencies=dependencies,
+    )
+    handoff_history = handoff_history_result.history
+
+    diagnostics.extend(
+        AnalysisDiagnostic(
+            code=diagnostic.code,
+            subject_id=f"{diagnostic.upstream_task}->{diagnostic.downstream_task}",
+            message=diagnostic.message,
+        )
+        for diagnostic in handoff_history_result.diagnostics
+    )
+
+    handoff_drift_results = {}
+
+    for edge, delays in handoff_history.items():
+        upstream_task, downstream_task = edge
+
+        try:
+            handoff_drift_results[edge] = calculate_handoff_drift(
+                upstream_task=upstream_task,
+                downstream_task=downstream_task,
+                handoff_delays=delays,
+                policy=policy,
+            )
+        except InsufficientHistoryError as exc:
+            diagnostics.append(
+                AnalysisDiagnostic(
+                    code="INSUFFICIENT_HANDOFF_HISTORY",
+                    subject_id=f"{upstream_task}->{downstream_task}",
+                    message=str(exc),
+                )
+            )
+
+    task_impacts = {}
+
+    for task_id, task_drift in drift_results.items():
+        upstream_handoff_drifts = [
+            drift
+            for (
+                _upstream_task,
+                downstream_task,
+            ), drift in handoff_drift_results.items()
+            if downstream_task == task_id
+        ]
+
+        task_impacts[task_id] = classify_task_impact(
+            task_id=task_id,
+            task_drift=task_drift,
+            upstream_handoff_drifts=upstream_handoff_drifts,
+        )
+
+    propagation_results = analyze_propagation(
+        drift_results=drift_results,
+        dependencies=dependencies,
+    )
+
+    overall_severity = Severity.NORMAL
+    all_drift_results = [
+        *drift_results.values(),
+        *handoff_drift_results.values(),
+    ]
+
+    if all_drift_results:
+        overall_severity = max(
+            all_drift_results,
+            key=lambda result: SEVERITY_SCORE[result.severity],
+        ).severity
+
+    primary_origin = select_primary_origin(
+        drift_results=drift_results,
+        task_impacts=task_impacts,
+        dependencies=dependencies,
+        propagation_results=propagation_results,
+    )
+
+    return DAGAnalysis(
+        dag_id=dag_id,
+        runs_analyzed=len({run.dag_run_id for run in task_runs}),
+        overall_severity=overall_severity,
+        primary_origin=primary_origin,
+        drift_results=drift_results,
+        handoff_drift_results=handoff_drift_results,
+        task_impacts=task_impacts,
+        propagation_results=propagation_results,
+        dependencies=dependencies,
+        diagnostics=diagnostics,
+        policy=policy,
+    )

src/flowsense/application/ports.py

@@ -0,0 +1,9 @@
+from typing import Protocol
+
+from flowsense.domain import TaskRun
+
+
+class DAGDataSource(Protocol):
+    def collect_task_runs(self, dag_id: str) -> list[TaskRun]: ...
+
+    def get_dag_dependencies(self, dag_id: str) -> dict[str, list[str]]: ...