diff --git a/docs/bugs.md b/docs/bugs.md new file mode 100644 index 0000000..9fa2c35 --- /dev/null +++ b/docs/bugs.md @@ -0,0 +1,213 @@ +This is a solid, well-structured repo. The architecture matches the spec perfectly and most of the code is production-quality. Here's the full audit — bugs first, then improvements. + +*** + +## 🔴 Bugs (Will Break at Runtime) + +### Bug 1 — `asyncio.run()` inside an already-running event loop crashes +**File:** `pipelineprobe/connectors/postgres.py` line `get_stats_sync()` + +`asyncio.run()` throws `RuntimeError: This event loop is already running` if PipelineProbe is ever called from an async context (FastAPI, Jupyter, any async test runner). This is a latent crash waiting to happen. + +**Fix:** Replace the async/sync split entirely. `asyncpg` is overkill for a one-shot audit query. Use `psycopg2` synchronously: + +```python +import logging +from typing import Any, Dict, List + +import psycopg2 +import psycopg2.extras + +from pipelineprobe.config import WarehouseConfig + +logger = logging.getLogger(__name__) + +class PostgresConnector: + def __init__(self, config: WarehouseConfig): + self.config = config + + def get_stats_sync(self) -> List[Dict[str, Any]]: + try: + conn = psycopg2.connect(self.config.dsn) + with conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) as cur: + cur.execute(""" + SELECT + t.schemaname, + t.relname AS tablename, + t.n_live_tup AS row_count, + EXISTS ( + SELECT 1 FROM information_schema.columns c + WHERE c.table_schema = t.schemaname + AND c.table_name = t.relname + AND c.column_name IN ('updated_at', 'created_at') + ) AS has_timestamps + FROM pg_stat_user_tables t + ORDER BY t.n_live_tup DESC + LIMIT 50 + """) + return [dict(r) for r in cur.fetchall()] + except Exception as e: + logger.error("Error connecting to Postgres: %s", e) + return [] + finally: + if conn: + conn.close() +``` + +Also update `pyproject.toml` — remove `asyncpg`, add `psycopg2-binary`. + +*** + +### Bug 2 — `config.py`: `AirflowConfig` field is `base_url` but `init` command writes `url` +**File:** `pipelineprobe/config.py` line 8 vs `cli.py` `init` command + +`AirflowConfig` defines `base_url` as the field name, but the `init` command generates a YAML stub with key `url`. When a user runs `pipelineprobe init` then `pipelineprobe audit`, the `base_url` field gets its default (`http://localhost:8080`) instead of their value because the YAML key doesn't match. + +**Fix in `cli.py` init command** — change the generated YAML: +```yaml +orchestrator: + base_url: "http://localhost:8080" # ← was "url:" + username: "admin" +``` + +*** + +### Bug 3 — `cli.py`: `postgres_tables` hardcoded context key doesn't work for BigQuery/Snowflake rules +**File:** `pipelineprobe/cli.py` line 66 + +```python +"postgres_tables": warehouse_tables, # Backwards compatible context key +``` + +The comment says "backwards compatible" but `postgres_rules.py` reads from `postgres_tables` key while BigQuery and Snowflake also return their data into the same key. This is fine for now — but the `postgres_rules.py` rule will fire on BigQuery/Snowflake data with misleading Postgres-specific recommendations (e.g. "consider partitioning" using Postgres terminology on a BigQuery table). + +**Fix:** Use a neutral key and pass warehouse type to the context: +```python +context = { + "airflow_dags": airflow_dags, + "airflow_tasks": airflow_tasks, + "dbt_models": dbt_models, + "warehouse_tables": warehouse_tables, # ← neutral key + "warehouse_type": cfg.warehouse.type, # ← pass type for rule conditions +} +``` +Update `postgres_rules.py` to read `warehouse_tables` and guard with `if context.get("warehouse_type") == "postgres"`. + +*** + +### Bug 4 — `dbt.py`: manifest path is double-joined +**File:** `pipelineprobe/connectors/dbt.py` lines 16–17 + +```python +manifest_path = project_dir / self.config.manifest_path +``` + +`DbtConfig` defaults: +```python +project_dir: str = "./analytics" +manifest_path: str = "target/manifest.json" +``` + +So the resolved path becomes `./analytics/target/manifest.json`. But many users set `project_dir` to the dbt project root AND `manifest_path` to a full relative path from CWD like `./analytics/target/manifest.json`. This causes a double-join. The config field name `manifest_path` implies it's already a full path, not relative to `project_dir`. + +**Fix:** Make `manifest_path` and `run_results_path` absolute-or-CWD-relative, not relative to `project_dir`: +```python +manifest_path = Path(self.config.manifest_path) +run_results_path = Path(self.config.run_results_path) +``` +Update defaults in `DbtConfig` to `"./analytics/target/manifest.json"` to match the original intent, and update the generated `init` YAML accordingly. + +*** + +### Bug 5 — `airflow_rules.py`: `check_stale_dags` uses `.days` which truncates — misses sub-day staleness +**File:** `pipelineprobe/rules/airflow_rules.py` line 93 + +```python +days_since = (now - end_time_aware).days +``` + +`timedelta.days` is the integer day component only — it **does not round**. A DAG that last succeeded 6 days and 23 hours ago returns `.days == 6`, not 7, so it never triggers the `> 7` check. Use `total_seconds()` instead: + +```python +days_since = (now - end_time_aware).total_seconds() / 86400 +if days_since > stale_threshold_days: +``` + +*** + +## 🟡 Issues That Will Cause Confusion + +### Issue 1 — `pyproject.toml` likely missing `asyncpg` / `python-dateutil` / `snowflake-connector-python` as explicit deps +The connectors import `asyncpg`, `dateutil`, and `snowflake.connector` but without seeing `pyproject.toml` contents fully, these are often missed. Verify the `[project.dependencies]` section includes: +```toml +dependencies = [ + "typer>=0.9", + "httpx>=0.27", + "pydantic>=2.0", + "pydantic-settings>=2.0", + "pyyaml>=6.0", + "jinja2>=3.1", + "python-dateutil>=2.8", + "psycopg2-binary>=2.9", # after fixing Bug 1 + "snowflake-connector-python>=3.0", + "google-cloud-bigquery>=3.0", +] +``` + +*** + +### Issue 2 — `cli.py`: `fail_on_critical` default is `5` in config but `0` in `init` YAML +`ReportConfig` defaults `fail_on_critical: int = 5` but the `init` command generates `fail_on_critical: 0`. A user who runs `init`, gets the YAML, and doesn't edit it will find the audit **always fails on any critical issue**. The defaults should be consistent — pick `5` in both places. + +*** + +### Issue 3 — `renderer.py`: `report.html` template missing — `TemplateNotFound` at runtime +The `renderer.py` loads `report.html` from `pipelineprobe/templates/` but the `templates/` directory exists — if `report.html` is empty or missing, every audit run fails at the last step with a Jinja2 `TemplateNotFound` error. Verify the template file exists and has actual HTML. If it's a stub, even a minimal one like this is enough to unblock users: + +```html + + +PipelineProbe Report + +

Score: {{ summary.score }}/100

+

Critical: {{ summary.critical_count }} | Warnings: {{ summary.warning_count }}

+ {% for issue in issues %} +
+ {{ issue.summary }} +

{{ issue.recommendation }}

+
+ {% endfor %} + + +``` + +*** + +## ✅ What's Already Good + +| Area | Verdict | +|------|---------| +| Overall architecture (connectors / rules / renderer) | Clean, matches spec exactly | +| Airflow pagination with `offset` loop | Correct | +| Snowflake CTE approach (avoids correlated subquery restriction) | Smart fix | +| dbt test count via `depends_on.nodes` traversal | Correct approach | +| Timezone-aware `datetime.now(timezone.utc)` in rules | Correct | +| `fail_on_critical` CLI override | Good UX | +| Partial report on connector failure (returns `[]`) | Resilient | +| Snowflake missing-credential guard | Correct | + +*** + +## Priority Fix Order + +``` +1. Bug 1 — Replace asyncpg/asyncio.run with psycopg2 sync (crash risk) +2. Bug 2 — Fix init YAML key url → base_url (silent misconfiguration) +3. Bug 3 — Neutral warehouse_tables context key (wrong rules on wrong warehouse) +4. Bug 4 — Fix dbt manifest double-join path (FileNotFoundError for all dbt users) +5. Bug 5 — Use total_seconds() for stale DAG check (off-by-<1-day logic error) +6. Issue 2 — Align fail_on_critical defaults (unexpected CI failures) +7. Issue 3 — Confirm report.html template exists and renders +``` + +Fix bugs 1–4 before any public share or PyPI publish — they will hit every first-time user. \ No newline at end of file diff --git a/pipelineprobe/cli.py b/pipelineprobe/cli.py index 93df6cc..3b20bc4 100644 --- a/pipelineprobe/cli.py +++ b/pipelineprobe/cli.py @@ -73,7 +73,8 @@ def audit( "airflow_dags": airflow_dags, "airflow_tasks": airflow_tasks, "dbt_models": dbt_models, - "postgres_tables": warehouse_tables, # Backwards compatible context key + "warehouse_tables": warehouse_tables, + "warehouse_type": cfg.warehouse.type, } typer.echo("Running rule engine...") @@ -123,13 +124,15 @@ def init(): default_config = """# PipelineProbe Configuration orchestrator: - url: "http://localhost:8080" + base_url: "http://localhost:8080" username: "admin" # Set PIPELINEPROBE_AIRFLOW_PASSWORD in environment instead of hardcoding here dbt: project_dir: "./dbt" target: "dev" + manifest_path: "./dbt/target/manifest.json" + run_results_path: "./dbt/target/run_results.json" warehouse: # Set PIPELINEPROBE_WAREHOUSE_DSN in environment @@ -138,7 +141,7 @@ def init(): report: output_dir: "./reports" format: "both" - fail_on_critical: 0 + fail_on_critical: 5 """ with open("pipelineprobe.yml", "w") as f: f.write(default_config) diff --git a/pipelineprobe/config.py b/pipelineprobe/config.py index dbcae1f..d9a2dc0 100644 --- a/pipelineprobe/config.py +++ b/pipelineprobe/config.py @@ -14,8 +14,8 @@ class AirflowConfig(BaseModel): class DbtConfig(BaseModel): project_dir: str = "./analytics" target: str = "prod" - manifest_path: str = "target/manifest.json" - run_results_path: str = "target/run_results.json" + manifest_path: str = "./analytics/target/manifest.json" + run_results_path: str = "./analytics/target/run_results.json" class WarehouseConfig(BaseModel): type: str = "postgres" diff --git a/pipelineprobe/connectors/dbt.py b/pipelineprobe/connectors/dbt.py index 2279246..2ce3dd5 100644 --- a/pipelineprobe/connectors/dbt.py +++ b/pipelineprobe/connectors/dbt.py @@ -14,8 +14,8 @@ def __init__(self, config: DbtConfig): def get_models(self) -> List[DbtModel]: project_dir = Path(self.config.project_dir) - manifest_path = project_dir / self.config.manifest_path - run_results_path = project_dir / self.config.run_results_path + manifest_path = Path(self.config.manifest_path) + 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) diff --git a/pipelineprobe/connectors/postgres.py b/pipelineprobe/connectors/postgres.py index c6ff5db..c303ab5 100644 --- a/pipelineprobe/connectors/postgres.py +++ b/pipelineprobe/connectors/postgres.py @@ -1,8 +1,8 @@ -import asyncio import logging from typing import Any, Dict, List -import asyncpg +import psycopg2 +import psycopg2.extras from pipelineprobe.config import WarehouseConfig @@ -12,34 +12,31 @@ class PostgresConnector: def __init__(self, config: WarehouseConfig): self.config = config - async def get_table_stats(self) -> List[Dict[str, Any]]: + def get_stats_sync(self) -> List[Dict[str, Any]]: conn = None try: - conn = await asyncpg.connect(self.config.dsn) - query = """ - SELECT - t.schemaname, - t.relname as tablename, - t.n_live_tup as row_count, - EXISTS ( - SELECT 1 - FROM information_schema.columns c - WHERE c.table_schema = t.schemaname - AND c.table_name = t.relname - AND c.column_name IN ('updated_at', 'created_at') - ) as has_timestamps - FROM pg_stat_user_tables t - ORDER BY t.n_live_tup DESC - LIMIT 50; - """ - rows = await conn.fetch(query) - return [dict(r) for r in rows] + conn = psycopg2.connect(self.config.dsn) + with conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) as cur: + cur.execute(""" + SELECT + t.schemaname, + t.relname AS tablename, + t.n_live_tup AS row_count, + EXISTS ( + SELECT 1 FROM information_schema.columns c + WHERE c.table_schema = t.schemaname + AND c.table_name = t.relname + AND c.column_name IN ('updated_at', 'created_at') + ) AS has_timestamps + FROM pg_stat_user_tables t + ORDER BY t.n_live_tup DESC + LIMIT 50 + """) + return [dict(r) for r in cur.fetchall()] except Exception as e: logger.error("Error connecting to Postgres: %s", e) return [] finally: if conn: - await conn.close() - - def get_stats_sync(self) -> List[Dict[str, Any]]: - return asyncio.run(self.get_table_stats()) + conn.close() + diff --git a/pipelineprobe/rules/airflow_rules.py b/pipelineprobe/rules/airflow_rules.py index 95ccb25..5cdaadb 100644 --- a/pipelineprobe/rules/airflow_rules.py +++ b/pipelineprobe/rules/airflow_rules.py @@ -96,7 +96,7 @@ def check_stale_dags(context: dict) -> List[Issue]: if end_time_aware.tzinfo is None: end_time_aware = end_time_aware.replace(tzinfo=timezone.utc) - days_since = (now - end_time_aware).days + days_since = (now - end_time_aware).total_seconds() / 86400 if days_since > stale_threshold_days: issues.append( Issue( diff --git a/pipelineprobe/rules/postgres_rules.py b/pipelineprobe/rules/postgres_rules.py index 8146443..8d6c360 100644 --- a/pipelineprobe/rules/postgres_rules.py +++ b/pipelineprobe/rules/postgres_rules.py @@ -3,7 +3,9 @@ def check_large_tables(context: dict) -> List[Issue]: issues = [] - tables: List[Dict[str, Any]] = context.get("postgres_tables", []) + if context.get("warehouse_type") != "postgres": + return issues + tables: List[Dict[str, Any]] = context.get("warehouse_tables", []) for table in tables: if table.get("row_count", 0) > 10_000_000: issues.append( @@ -20,7 +22,9 @@ def check_large_tables(context: dict) -> List[Issue]: def check_missing_timestamps(context: dict) -> List[Issue]: issues = [] - tables: List[Dict[str, Any]] = context.get("postgres_tables", []) + if context.get("warehouse_type") != "postgres": + return issues + tables: List[Dict[str, Any]] = context.get("warehouse_tables", []) for table in tables: # For simplicity, we flag tables > 1,000,000 rows without timestamps if table.get("row_count", 0) > 1_000_000 and not table.get("has_timestamps"): diff --git a/pyproject.toml b/pyproject.toml index ce1a03b..920efdf 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -18,7 +18,7 @@ dependencies = [ "pydantic-settings>=2.0.0", "httpx>=0.25.0", "jinja2>=3.1.2", - "asyncpg>=0.28.0", + "psycopg2-binary>=2.9.0", "pyyaml>=6.0", "python-dateutil>=2.8.0", "google-cloud-bigquery>=3.11.0", diff --git a/tests/test_connectors.py b/tests/test_connectors.py index 13d0568..9ea9d24 100644 --- a/tests/test_connectors.py +++ b/tests/test_connectors.py @@ -9,29 +9,27 @@ from pipelineprobe.connectors.bigquery import BigQueryConnector from pipelineprobe.connectors.airflow import AirflowConnector -def test_postgres_connector(): +@patch("pipelineprobe.connectors.postgres.psycopg2.connect") +def test_postgres_connector(mock_connect): config = WarehouseConfig(type="postgres", dsn="postgresql://fake") connector = PostgresConnector(config) - with patch("asyncpg.connect", new_callable=AsyncMock) as mock_connect: - mock_conn = AsyncMock() - # Mocks fetch returning a list of dict-like records - mock_conn.fetch.return_value = [{"schemaname": "public", "tablename": "users", "row_count": 100, "has_timestamps": True}] - mock_connect.return_value = mock_conn - - # Test sync wrapper - stats_sync = connector.get_stats_sync() - assert len(stats_sync) == 1 - assert stats_sync[0]["has_timestamps"] is True + # Mock context manager cursor + mock_cursor = mock_connect.return_value.cursor.return_value.__enter__.return_value + mock_cursor.fetchall.return_value = [{"schemaname": "public", "tablename": "users", "row_count": 100, "has_timestamps": True}] + + stats_sync = connector.get_stats_sync() + assert len(stats_sync) == 1 + assert stats_sync[0]["has_timestamps"] is True -def test_postgres_connector_error(): +@patch("pipelineprobe.connectors.postgres.psycopg2.connect") +def test_postgres_connector_error(mock_connect): config = WarehouseConfig(type="postgres", dsn="postgresql://fake") connector = PostgresConnector(config) - with patch("asyncpg.connect", new_callable=AsyncMock) as mock_connect: - mock_connect.side_effect = Exception("DB Down") - stats = connector.get_stats_sync() - assert stats == [] + mock_connect.side_effect = Exception("DB Down") + stats = connector.get_stats_sync() + assert stats == [] def test_snowflake_connector_missing_creds(): # Snowflake connector should validate account, username, password @@ -128,7 +126,7 @@ def test_dbt_connector_success(tmp_path): rr_path = tmp_path / "rr.json" rr_path.write_text(json.dumps(rr)) - config = DbtConfig(project_dir=str(tmp_path), manifest_path="manifest.json", run_results_path="rr.json") + config = DbtConfig(project_dir=str(tmp_path), manifest_path=str(manifest_path), run_results_path=str(rr_path)) connector = DbtConnector(config) models = connector.get_models() diff --git a/tests/test_rules.py b/tests/test_rules.py index fe34651..a29fef3 100644 --- a/tests/test_rules.py +++ b/tests/test_rules.py @@ -58,7 +58,10 @@ def test_check_high_failure_rate(): assert "failing_dag" in issues[0].affected_resources def test_check_large_tables(mock_postgres_tables): - context = {"postgres_tables": mock_postgres_tables} + context = { + "warehouse_tables": mock_postgres_tables, + "warehouse_type": "postgres" + } issues = check_large_tables(context) assert len(issues) == 1 @@ -72,7 +75,10 @@ def test_check_missing_timestamps(): {"tablename": "huge_with_ts", "row_count": 5_000_000, "has_timestamps": True}, {"tablename": "small_no_ts", "row_count": 100, "has_timestamps": False} ] - issues = check_missing_timestamps({"postgres_tables": tables}) + issues = check_missing_timestamps({ + "warehouse_tables": tables, + "warehouse_type": "postgres" + }) assert len(issues) == 1 assert "huge_no_ts" in issues[0].affected_resources