style: apply ruff format to unify codebase style

.vscode/settings.json

@@ -0,0 +1 @@
+{}

pipelineprobe/__init__.py

@@ -1,4 +1,5 @@
 """
 PipelineProbe: Instant Data Pipeline Audit Report
 """
+
 __version__ = "0.1.0"

pipelineprobe/cli.py

@@ -18,32 +18,40 @@
     add_completion=False,
 )
 
+
 @app.command()
 def audit(
     config: str = typer.Option("pipelineprobe.yml", help="Path to config file"),
-    fail_on_critical: int = typer.Option(None, help="Override fail-on-critical threshold"),
-    report_format: str = typer.Option(None, "--format", help="Override report format: html | json | both")
+    fail_on_critical: int = typer.Option(
+        None, help="Override fail-on-critical threshold"
+    ),
+    report_format: str = typer.Option(
+        None, "--format", help="Override report format: html | json | both"
+    ),
 ):
     """
     Run the PipelineProbe audit and generate a report.
     """
     typer.echo(f"Loading configuration from {config}...")
     cfg = load_config(config)
-    
+
     # Apply CLI overrides
     if fail_on_critical is not None:
         cfg.report.fail_on_critical = fail_on_critical
     if report_format is not None:
         valid_formats = {"html", "json", "both"}
         if report_format not in valid_formats:
-            typer.secho(f"Invalid --format '{report_format}'. Must be one of: {', '.join(sorted(valid_formats))}", fg=typer.colors.RED)
+            typer.secho(
+                f"Invalid --format '{report_format}'. Must be one of: {', '.join(sorted(valid_formats))}",
+                fg=typer.colors.RED,
+            )
             raise typer.Exit(code=1)
         cfg.report.format = report_format
 
     typer.echo("Initializing connectors...")
     airflow_conn = AirflowConnector(cfg.orchestrator)
     dbt_conn = DbtConnector(cfg.dbt)
-    
+
     if cfg.warehouse.type == "bigquery":
         typer.echo("Using BigQuery connector...")
         warehouse_conn = BigQueryConnector(cfg.warehouse)
@@ -58,14 +66,16 @@ def audit(
     # Fetch top-level DAGs
     airflow_dags = airflow_conn.get_dags()
     airflow_tasks = []
-    
+
     # Wire the actual dag runs and tasks for every returned DAG
     if airflow_dags:
-        typer.echo(f"Found {len(airflow_dags)} Airflow DAGs. Fetching runs and tasks...")
+        typer.echo(
+            f"Found {len(airflow_dags)} Airflow DAGs. Fetching runs and tasks..."
+        )
         for dag in airflow_dags:
             dag.recent_runs = airflow_conn.get_dag_runs(dag.id)
             airflow_tasks.extend(airflow_conn.get_tasks(dag.id))
-        
+
     dbt_models = dbt_conn.get_models()
     warehouse_tables = warehouse_conn.get_stats_sync()
 
@@ -90,28 +100,30 @@ def audit(
         "score": score,
         "critical_count": critical_count,
         "warning_count": warning_count,
-        "total_issues": len(issues)
+        "total_issues": len(issues),
     }
 
     typer.echo("Rendering reports...")
     renderer = ReportRenderer(cfg.report.output_dir)
-    
+
     if cfg.report.format in ("html", "both"):
         html_path = renderer.render_html(issues, summary)
         typer.echo(f"HTML report generated at: {html_path}")
-        
+
     if cfg.report.format in ("json", "both"):
         json_path = renderer.render_json(issues, summary)
         typer.echo(f"JSON report generated at: {json_path}")
-        
+
     if critical_count > cfg.report.fail_on_critical:
-        typer.secho(f"Audit failed! Found {critical_count} critical issues (threshold: {cfg.report.fail_on_critical}).", fg=typer.colors.RED)
+        typer.secho(
+            f"Audit failed! Found {critical_count} critical issues (threshold: {cfg.report.fail_on_critical}).",
+            fg=typer.colors.RED,
+        )
         raise typer.Exit(code=1)
 
     typer.secho("Audit completed successfully!", fg=typer.colors.GREEN)
 
 
-
 @app.command()
 def init():
     """
@@ -145,7 +157,7 @@ def init():
 """
     with open("pipelineprobe.yml", "w") as f:
         f.write(default_config)
-    
+
     typer.secho("Initialized pipelineprobe.yml successfully.", fg=typer.colors.GREEN)
 
 

pipelineprobe/config.py

@@ -3,6 +3,7 @@
 from pydantic import BaseModel
 from pydantic_settings import BaseSettings
 
+
 class AirflowConfig(BaseModel):
     type: str = "airflow"
     base_url: str = "http://localhost:8080"
@@ -11,12 +12,14 @@ class AirflowConfig(BaseModel):
     verify_ssl: bool = False
     lookback_days: int = 14
 
+
 class DbtConfig(BaseModel):
     project_dir: str = "./analytics"
     target: str = "prod"
     manifest_path: str = "./analytics/target/manifest.json"
     run_results_path: str = "./analytics/target/run_results.json"
 
+
 class WarehouseConfig(BaseModel):
     type: str = "postgres"
     dsn: str = "postgresql://user:pass@localhost:5432/analytics"
@@ -25,18 +28,21 @@ class WarehouseConfig(BaseModel):
     username: str | None = None
     password: str | None = None
 
+
 class ReportConfig(BaseModel):
     output_dir: str = "./reports"
     format: str = "html"
     include_cost_section: bool = False
     fail_on_critical: int = 5
 
+
 class PipelineProbeConfig(BaseSettings):
     orchestrator: AirflowConfig = AirflowConfig()
     dbt: DbtConfig = DbtConfig()
     warehouse: WarehouseConfig = WarehouseConfig()
     report: ReportConfig = ReportConfig()
 
+
 def load_config(config_path: str) -> PipelineProbeConfig:
     """Load config from YAML if it exists, otherwise return defaults."""
     path = Path(config_path)

pipelineprobe/connectors/airflow.py

@@ -16,15 +16,19 @@
 class AirflowConnector:
     def __init__(self, config: AirflowConfig):
         self.config = config
-        
+
         if not self.config.username or not self.config.password:
             logger.warning(
                 "Airflow credentials missing. Use PIPELINEPROBE_ORCHESTRATOR_USERNAME/PASSWORD "
                 "or update pipelineprobe.yml."
             )
-
-        auth = (self.config.username, self.config.password) if self.config.username and self.config.password else None
-
+
+        auth = (
+            (self.config.username, self.config.password)
+            if self.config.username and self.config.password
+            else None
+        )
+
         self.client = httpx.Client(
             base_url=self.config.base_url,
             auth=auth,
@@ -76,9 +80,15 @@ def get_dag_runs(self, dag_id: str) -> List[DagRun]:
                 start_str = r.get("start_date")
                 end_str = r.get("end_date")
                 # Fall back to execution_date if start_date missing (older Airflow versions)
-                start_time = parse_date(start_str) if start_str else parse_date(r["execution_date"])
+                start_time = (
+                    parse_date(start_str)
+                    if start_str
+                    else parse_date(r["execution_date"])
+                )
                 end_time = parse_date(end_str) if end_str else None
-                runs.append(DagRun(state=state, start_time=start_time, end_time=end_time))
+                runs.append(
+                    DagRun(state=state, start_time=start_time, end_time=end_time)
+                )
             return runs
         except Exception as e:
             logger.error("Error fetching DAG runs for %s: %s", dag_id, e)

pipelineprobe/connectors/dbt.py

@@ -8,6 +8,7 @@
 
 logger = logging.getLogger(__name__)
 
+
 class DbtConnector:
     def __init__(self, config: DbtConfig):
         self.config = config
@@ -17,7 +18,9 @@ def get_models(self) -> List[DbtModel]:
         run_results_path = Path(self.config.run_results_path)
 
         if not manifest_path.exists():
-            logger.warning("dbt manifest not found at %s — skipping dbt checks.", manifest_path)
+            logger.warning(
+                "dbt manifest not found at %s — skipping dbt checks.", manifest_path
+            )
             return []
 
         try:

pipelineprobe/connectors/postgres.py

@@ -8,6 +8,7 @@
 
 logger = logging.getLogger(__name__)
 
+
 class PostgresConnector:
     def __init__(self, config: WarehouseConfig):
         self.config = config
@@ -39,4 +40,3 @@ def get_stats_sync(self) -> List[Dict[str, Any]]:
         finally:
             if conn:
                 conn.close()
-

pipelineprobe/models.py

@@ -3,30 +3,35 @@
 
 from pydantic import BaseModel
 
+
 class DagRun(BaseModel):
     state: str
     start_time: datetime
     end_time: datetime | None = None
 
+
 class Dag(BaseModel):
     id: str
     is_active: bool
     recent_runs: list[DagRun]
     owner: str | None = None
 
+
 class Task(BaseModel):
     dag_id: str
     task_id: str
     retries: int
     sla: timedelta | None = None
     has_alerts: bool
 
+
 class DbtModel(BaseModel):
     name: str
     tests_count: int
     last_run_status: str
     tags: list[str]
 
+
 class Issue(BaseModel):
     severity: Literal["critical", "warning", "info"]
     category: Literal["dag", "task", "dbt", "warehouse"]

pipelineprobe/renderer.py

@@ -13,6 +13,7 @@ def _json_default(obj):
         return obj.total_seconds()
     raise TypeError(f"Object of type {type(obj).__name__} is not JSON serializable")
 
+
 class ReportRenderer:
     def __init__(self, output_dir: str):
         self.output_dir = Path(output_dir)
@@ -30,10 +31,7 @@ def render_html(self, issues: list[Issue], summary: dict) -> Path:
 
     def render_json(self, issues: list[Issue], summary: dict) -> Path:
         output_path = self.output_dir / "report.json"
-        data = {
-            "summary": summary,
-            "issues": [i.model_dump() for i in issues]
-        }
+        data = {"summary": summary, "issues": [i.model_dump() for i in issues]}
         with open(output_path, "w") as f:
             json.dump(data, f, indent=2, default=_json_default)
         return output_path

pipelineprobe/rules/__init__.py

@@ -3,6 +3,7 @@
 from .dbt_rules import register_dbt_rules
 from .postgres_rules import register_postgres_rules
 
+
 def get_configured_engine() -> RuleEngine:
     engine = RuleEngine()
     register_airflow_rules(engine)