diff --git a/src/energex/observer/config.py b/src/energex/observer/config.py index 50cffc8..04d1643 100644 --- a/src/energex/observer/config.py +++ b/src/energex/observer/config.py @@ -9,5 +9,12 @@ class ObserverSettings(BaseSettings): jwt_secret: SecretStr = Field(validation_alias="OBSERVER_JWT_SECRET") jwt_audience: str = Field(default="authenticated", validation_alias="OBSERVER_JWT_AUDIENCE") + supabase_url: str | None = Field(default=None, validation_alias="SUPABASE_URL") + supabase_service_key: SecretStr | None = Field( + default=None, validation_alias="SUPABASE_SERVICE_KEY" + ) + dagster_graphql_url: str = Field( + default="http://dagster-webserver:3000/graphql", validation_alias="DAGSTER_GRAPHQL_URL" + ) model_config = SettingsConfigDict(case_sensitive=False) diff --git a/src/energex/observer/dagster_client.py b/src/energex/observer/dagster_client.py new file mode 100644 index 0000000..7428689 --- /dev/null +++ b/src/energex/observer/dagster_client.py @@ -0,0 +1,189 @@ +"""Read-only Dagster GraphQL client for the Quality board. Sync httpx (routers are sync). + +Resilient: any transport/parse error -> {available: False} (the board degrades, never 500s). + +Query design confirmed against the running dagster-webserver (Dagster 1.13.9) during the task-1 +schema spike. Key deviations from the brief's assumed schema: + + - assetCheckExecutions(assetKey, checkName) is per-check, requires NON_NULL args — unusable for + a bulk scan. Instead we query assetNodes with embedded + assetChecksOrError.checks.executionForLatestMaterialization (one round-trip, all checks). + - MetadataEntry is an interface; concrete subtypes expose intValue/floatValue/text rather than a + generic `value` field. + - schedulesOrError requires repositorySelector (NON_NULL). Instead we use workspaceOrError and + collect schedules from embedded repositories. + - Check status enum values: SUCCEEDED / EXECUTION_FAILED / SKIPPED / FAILED (not PASSED). + passed = (status == "SUCCEEDED"). +""" + +from __future__ import annotations + +import os +import time + +import httpx + +# Single query that fetches all three data categories in one round-trip. +_CHECKS_QUERY = """ +query ObserverChecks { + assetNodes(loadMaterializations: false) { + assetKey { path } + assetChecksOrError { + __typename + ... on AssetChecks { + checks { + name + executionForLatestMaterialization { + runId status timestamp + evaluation { + metadataEntries { + label + ... on IntMetadataEntry { intValue } + ... on FloatMetadataEntry { floatValue } + ... on TextMetadataEntry { text } + } + } + } + } + } + } + } + runsOrError(limit: 100) { + ... on Runs { + results { id status startTime endTime assetSelection { path } } + } + } + workspaceOrError { + ... on Workspace { + locationEntries { + locationOrLoadError { + ... on RepositoryLocation { + repositories { + schedules { name scheduleState { status } } + } + } + } + } + } + } +} +""" + +_CACHE: dict[str, tuple[float, dict]] = {} +_TTL = 300 # seconds; override with OBSERVER_DAGSTER_TTL_SECONDS env var + + +def _dagster_url() -> str: + """Return the Dagster GraphQL URL from env (or the default) without requiring jwt_secret.""" + return os.environ.get("DAGSTER_GRAPHQL_URL", "http://dagster-webserver:3000/graphql") + + +def _query(url: str) -> dict: + with httpx.Client(timeout=10.0) as client: + resp = client.post(url, json={"query": _CHECKS_QUERY}) + resp.raise_for_status() + return resp.json() + + +def _metadata_value(entry: dict) -> object: + """Extract the concrete value from a MetadataEntry inline-fragment result.""" + if "intValue" in entry: + return entry["intValue"] + if "floatValue" in entry: + return entry["floatValue"] + if "text" in entry: + return entry["text"] + return None + + +def _parse(raw: dict) -> dict: + data = (raw or {}).get("data") or {} + + # --- checks (from assetNodes) --- + checks = [] + for node in data.get("assetNodes") or []: + asset_path = (node.get("assetKey") or {}).get("path") or [] + asset_name = asset_path[-1] if asset_path else None + checks_or_err = node.get("assetChecksOrError") or {} + if checks_or_err.get("__typename") != "AssetChecks": + continue + for chk in checks_or_err.get("checks") or []: + exec_ = chk.get("executionForLatestMaterialization") + if exec_ is None: + checks.append( + { + "name": chk.get("name"), + "asset": asset_name, + "passed": None, + "status": None, + "timestamp": None, + "metadata": {}, + "last_run_status": None, + } + ) + continue + status = exec_.get("status") + metadata_entries = (exec_.get("evaluation") or {}).get("metadataEntries") or [] + checks.append( + { + "name": chk.get("name"), + "asset": asset_name, + "passed": status == "SUCCEEDED", + "status": status, + "timestamp": exec_.get("timestamp"), + "metadata": { + e["label"]: _metadata_value(e) for e in metadata_entries if "label" in e + }, + "last_run_status": status, + } + ) + + # --- runs --- + runs_data = data.get("runsOrError") or {} + runs = [ + { + "id": r.get("id"), + "status": r.get("status"), + "startTime": r.get("startTime"), + "endTime": r.get("endTime"), + "assets": [".".join(a.get("path") or []) for a in (r.get("assetSelection") or [])], + } + for r in (runs_data.get("results") or []) + ] + + # --- schedules (collected from workspace → locations → repos) --- + schedules = [] + workspace = data.get("workspaceOrError") or {} + for entry in workspace.get("locationEntries") or []: + loc = entry.get("locationOrLoadError") or {} + for repo in loc.get("repositories") or []: + for sched in repo.get("schedules") or []: + schedules.append( + { + "name": sched.get("name"), + "status": (sched.get("scheduleState") or {}).get("status"), + "last_tick": None, + } + ) + + return { + "available": True, + "checks": checks, + "runs": runs, + "schedules": schedules, + "error": None, + } + + +def get_pipeline_health() -> dict: + """Return pipeline health snapshot. Never raises — any failure → available=False.""" + ttl = int(os.environ.get("OBSERVER_DAGSTER_TTL_SECONDS", _TTL)) + hit = _CACHE.get("health") + if hit and (time.monotonic() - hit[0]) < ttl: + return hit[1] + try: + result = _parse(_query(_dagster_url())) + except Exception as exc: # transport, HTTP status, or parse — degrade, never raise + result = {"available": False, "checks": [], "runs": [], "schedules": [], "error": str(exc)} + _CACHE["health"] = (time.monotonic(), result) + return result diff --git a/src/energex/observer/supabase_client.py b/src/energex/observer/supabase_client.py new file mode 100644 index 0000000..a427fec --- /dev/null +++ b/src/energex/observer/supabase_client.py @@ -0,0 +1,74 @@ +"""Supabase PostgREST client (service key), used server-side AFTER the router enforces the role. +The service key never reaches the browser. Sync httpx (routers are sync). + +Reads SUPABASE_URL and SUPABASE_SERVICE_KEY directly from the environment to avoid +requiring OBSERVER_JWT_SECRET when the module is used in isolation.""" + +from __future__ import annotations + +import os + +import httpx + + +class SupabaseError(RuntimeError): + pass + + +def _url() -> str | None: + return os.environ.get("SUPABASE_URL") + + +def _key() -> str | None: + return os.environ.get("SUPABASE_SERVICE_KEY") + + +def is_configured() -> bool: + return bool(_url() and _key()) + + +def _headers() -> dict: + key = _key() + return { + "apikey": key, + "Authorization": f"Bearer {key}", + "Content-Type": "application/json", + "Prefer": "return=representation", + } + + +def _request(method: str, table: str, *, json=None, params=None) -> httpx.Response: + url_base = _url() + if not (url_base and _key()): + raise SupabaseError("Supabase not configured (set SUPABASE_URL + SUPABASE_SERVICE_KEY)") + url = f"{url_base}/rest/v1/{table}" + try: + with httpx.Client(timeout=10.0) as client: + resp = client.request(method, url, json=json, params=params, headers=_headers()) + resp.raise_for_status() + return resp + except httpx.HTTPError as exc: + raise SupabaseError(f"supabase {method} {table}: {exc}") from exc + + +def read_issue_acks() -> list[dict]: + resp = _request("GET", "issue_acks", params={"select": "*", "order": "created_at.desc"}) + return resp.json() + + +def write_issue_ack(user_id: str, issue_key: str, status: str, note: str | None = None) -> dict: + resp = _request( + "POST", + "issue_acks", + json={"issue_key": issue_key, "user_id": user_id, "status": status, "note": note}, + ) + rows = resp.json() + return rows[0] if isinstance(rows, list) and rows else {} + + +def write_audit_log(action: str, target: str, detail: dict, user_id: str) -> None: + _request( + "POST", + "audit_log", + json={"action": action, "target": target, "detail": detail or {}, "user_id": user_id}, + ) diff --git a/tests/test_observer_dagster_client.py b/tests/test_observer_dagster_client.py new file mode 100644 index 0000000..deb31c2 --- /dev/null +++ b/tests/test_observer_dagster_client.py @@ -0,0 +1,262 @@ +"""Tests for the read-only Dagster GraphQL client. + +The live schema (Dagster 1.13.9) differs from what the brief assumed: +- assetCheckExecutions requires assetKey+checkName (NON_NULL), returns a list directly (no .nodes). +- MetadataEntry is an interface; concrete types expose intValue/floatValue/text, not a generic `value`. +- schedulesOrError requires repositorySelector (NON_NULL). + +The client instead uses: +- assetNodes with embedded assetChecksOrError.checks.executionForLatestMaterialization +- runsOrError (unchanged) +- workspaceOrError to collect schedules without needing repositorySelector + +All tests mock httpx.Client.post so they run fully offline. +""" + +from __future__ import annotations + +import httpx + +from energex.observer import dagster_client + + +def _fake_post(payload): + def _post(self, url, json=None, **kw): + return httpx.Response(200, json=payload, request=httpx.Request("POST", url)) + + return _post + + +# --- realistic payload shape (live Dagster 1.13.9 schema) --- + +_HEALTHY_PAYLOAD = { + "data": { + "assetNodes": [ + { + "assetKey": {"path": ["ercot_load"]}, + "assetChecksOrError": { + "__typename": "AssetChecks", + "checks": [ + { + "name": "ercot_load_pass_quality_gate", + "executionForLatestMaterialization": { + "runId": "r1", + "status": "SUCCEEDED", + "timestamp": 1.0, + "evaluation": { + "metadataEntries": [ + {"label": "rows", "intValue": 42}, + {"label": "symbols", "intValue": 5}, + ] + }, + }, + } + ], + }, + }, + { + "assetKey": {"path": ["eia930_region"]}, + "assetChecksOrError": { + "__typename": "AssetChecks", + "checks": [ + { + "name": "eia930_region_pass_quality_gate", + "executionForLatestMaterialization": { + "runId": "r2", + "status": "EXECUTION_FAILED", + "timestamp": 2.0, + "evaluation": None, + }, + } + ], + }, + }, + # asset with no checks — must not crash + { + "assetKey": {"path": ["intraday_futures_bars"]}, + "assetChecksOrError": { + "__typename": "AssetChecks", + "checks": [], + }, + }, + ], + "runsOrError": { + "results": [ + { + "id": "r1", + "status": "SUCCESS", + "startTime": 1.0, + "endTime": 2.0, + "assetSelection": None, + } + ] + }, + "workspaceOrError": { + "__typename": "Workspace", + "locationEntries": [ + { + "locationOrLoadError": { + "__typename": "RepositoryLocation", + "repositories": [ + { + "schedules": [ + { + "name": "ercot_load_schedule", + "scheduleState": {"status": "RUNNING"}, + } + ] + } + ], + } + } + ], + }, + } +} + + +def test_get_pipeline_health_parses_checks(monkeypatch): + monkeypatch.setattr(httpx.Client, "post", _fake_post(_HEALTHY_PAYLOAD)) + dagster_client._CACHE.clear() + + health = dagster_client.get_pipeline_health() + + assert health["available"] is True + + chk = next(c for c in health["checks"] if c["name"] == "ercot_load_pass_quality_gate") + assert chk["passed"] is True + assert chk["asset"] == "ercot_load" + assert chk["metadata"]["rows"] == 42 + + failed_chk = next(c for c in health["checks"] if c["name"] == "eia930_region_pass_quality_gate") + assert failed_chk["passed"] is False + + +def test_get_pipeline_health_parses_runs(monkeypatch): + monkeypatch.setattr(httpx.Client, "post", _fake_post(_HEALTHY_PAYLOAD)) + dagster_client._CACHE.clear() + + health = dagster_client.get_pipeline_health() + + assert len(health["runs"]) == 1 + assert health["runs"][0]["id"] == "r1" + assert health["runs"][0]["status"] == "SUCCESS" + + +def test_get_pipeline_health_parses_schedules(monkeypatch): + monkeypatch.setattr(httpx.Client, "post", _fake_post(_HEALTHY_PAYLOAD)) + dagster_client._CACHE.clear() + + health = dagster_client.get_pipeline_health() + + assert len(health["schedules"]) == 1 + assert health["schedules"][0]["name"] == "ercot_load_schedule" + assert health["schedules"][0]["status"] == "RUNNING" + + +def test_get_pipeline_health_degrades_when_dagster_down(monkeypatch): + def _boom(self, *a, **k): + raise httpx.ConnectError("refused") + + monkeypatch.setattr(httpx.Client, "post", _boom) + dagster_client._CACHE.clear() + + health = dagster_client.get_pipeline_health() + + assert health["available"] is False + assert health["checks"] == [] + assert health["runs"] == [] + assert health["schedules"] == [] + assert health["error"] + + +def test_get_pipeline_health_degrades_on_http_error(monkeypatch): + def _bad(self, url, json=None, **kw): + return httpx.Response(500, text="Internal Server Error", request=httpx.Request("POST", url)) + + monkeypatch.setattr(httpx.Client, "post", _bad) + dagster_client._CACHE.clear() + + health = dagster_client.get_pipeline_health() + + assert health["available"] is False + assert health["error"] + + +def test_get_pipeline_health_degrades_on_missing_fields(monkeypatch): + """Partial/empty payload must not KeyError — degrades to empty lists.""" + monkeypatch.setattr(httpx.Client, "post", _fake_post({"data": {}})) + dagster_client._CACHE.clear() + + health = dagster_client.get_pipeline_health() + + assert health["available"] is True + assert health["checks"] == [] + assert health["runs"] == [] + assert health["schedules"] == [] + assert health["error"] is None + + +def test_get_pipeline_health_skips_asset_nodes_without_checks(monkeypatch): + """assetNodes with no checks must produce zero check entries, not crash.""" + monkeypatch.setattr(httpx.Client, "post", _fake_post(_HEALTHY_PAYLOAD)) + dagster_client._CACHE.clear() + + health = dagster_client.get_pipeline_health() + + # intraday_futures_bars has empty checks list — confirm it contributed nothing + check_assets = {c["asset"] for c in health["checks"]} + assert "intraday_futures_bars" not in check_assets + + +def test_dagster_client_import_is_safe(): + """Importing dagster_client must not raise even without env vars set.""" + import importlib + + importlib.reload(dagster_client) + assert hasattr(dagster_client, "get_pipeline_health") + + +def test_cache_returns_stale_result_within_ttl(monkeypatch): + """Second call within TTL must be served from cache without hitting the network.""" + monkeypatch.setattr(httpx.Client, "post", _fake_post(_HEALTHY_PAYLOAD)) + dagster_client._CACHE.clear() + + first = dagster_client.get_pipeline_health() + assert first["available"] is True + + def _boom(self, *a, **k): + raise httpx.ConnectError("must not be called within TTL") + + monkeypatch.setattr(httpx.Client, "post", _boom) + second = dagster_client.get_pipeline_health() + assert second["available"] is True # served from cache, no raise + + +def test_parse_data_null_degrades_safely(monkeypatch): + """A {data: null, errors: [...]} response must not raise and returns normalized shape.""" + payload = {"data": None, "errors": [{"message": "something failed"}]} + monkeypatch.setattr(httpx.Client, "post", _fake_post(payload)) + dagster_client._CACHE.clear() + + health = dagster_client.get_pipeline_health() + + assert health["available"] is True + assert health["checks"] == [] + assert health["runs"] == [] + assert health["schedules"] == [] + assert health["error"] is None + + +def test_config_accepts_missing_supabase_vars(monkeypatch): + """ObserverSettings must construct without SUPABASE_URL / SUPABASE_SERVICE_KEY.""" + monkeypatch.setenv("OBSERVER_JWT_SECRET", "test-secret") + monkeypatch.delenv("SUPABASE_URL", raising=False) + monkeypatch.delenv("SUPABASE_SERVICE_KEY", raising=False) + + from energex.observer.config import ObserverSettings + + s = ObserverSettings() + assert s.supabase_url is None + assert s.supabase_service_key is None + assert s.dagster_graphql_url == "http://dagster-webserver:3000/graphql" diff --git a/tests/test_observer_supabase_client.py b/tests/test_observer_supabase_client.py new file mode 100644 index 0000000..0d02a43 --- /dev/null +++ b/tests/test_observer_supabase_client.py @@ -0,0 +1,111 @@ +import httpx +import pytest + +from energex.observer import supabase_client as sc + + +def _capture(calls, status=201, body=None): + def _req(self, method, url, **kw): + calls.append((method, url, kw.get("json"), kw.get("headers"))) + return httpx.Response( + status, + json=body if body is not None else [{"id": 1}], + request=httpx.Request(method, url), + ) + + return _req + + +def test_write_issue_ack_posts_to_postgrest(monkeypatch): + monkeypatch.setenv("SUPABASE_URL", "http://sb.local") + monkeypatch.setenv("SUPABASE_SERVICE_KEY", "svc-key") + calls = [] + monkeypatch.setattr(httpx.Client, "request", _capture(calls)) + sc.write_issue_ack("uid-1", "check:ercot_load_pass_quality_gate", "ack", "looks transient") + method, url, body, headers = calls[0] + assert method == "POST" and url.endswith("/rest/v1/issue_acks") + assert body == { + "issue_key": "check:ercot_load_pass_quality_gate", + "user_id": "uid-1", + "status": "ack", + "note": "looks transient", + } + assert headers["apikey"] == "svc-key" and headers["Authorization"] == "Bearer svc-key" + + +def test_not_configured_raises(monkeypatch): + monkeypatch.delenv("SUPABASE_URL", raising=False) + monkeypatch.delenv("SUPABASE_SERVICE_KEY", raising=False) + assert sc.is_configured() is False + with pytest.raises(sc.SupabaseError): + sc.write_audit_log("ack_issue", "check:x", {}, "uid-1") + + +def test_write_audit_log_posts_to_postgrest(monkeypatch): + monkeypatch.setenv("SUPABASE_URL", "http://sb.local") + monkeypatch.setenv("SUPABASE_SERVICE_KEY", "svc-key") + calls = [] + monkeypatch.setattr(httpx.Client, "request", _capture(calls, status=201, body=[])) + result = sc.write_audit_log("ack_issue", "check:x", {"foo": "bar"}, "uid-2") + assert result is None + method, url, body, headers = calls[0] + assert method == "POST" and url.endswith("/rest/v1/audit_log") + assert body == { + "action": "ack_issue", + "target": "check:x", + "detail": {"foo": "bar"}, + "user_id": "uid-2", + } + + +def test_read_issue_acks_gets_with_params(monkeypatch): + monkeypatch.setenv("SUPABASE_URL", "http://sb.local") + monkeypatch.setenv("SUPABASE_SERVICE_KEY", "svc-key") + calls = [] + monkeypatch.setattr( + httpx.Client, + "request", + _capture(calls, status=200, body=[{"id": 1, "status": "ack"}]), + ) + rows = sc.read_issue_acks() + method, url, body, headers = calls[0] + assert method == "GET" and url.endswith("/rest/v1/issue_acks") + assert rows == [{"id": 1, "status": "ack"}] + + +def test_write_issue_ack_returns_row(monkeypatch): + monkeypatch.setenv("SUPABASE_URL", "http://sb.local") + monkeypatch.setenv("SUPABASE_SERVICE_KEY", "svc-key") + monkeypatch.setattr( + httpx.Client, + "request", + _capture([], status=201, body=[{"id": 99, "status": "ack"}]), + ) + row = sc.write_issue_ack("u", "k", "ack") + assert row == {"id": 99, "status": "ack"} + + +def test_write_issue_ack_empty_response_returns_empty_dict(monkeypatch): + monkeypatch.setenv("SUPABASE_URL", "http://sb.local") + monkeypatch.setenv("SUPABASE_SERVICE_KEY", "svc-key") + monkeypatch.setattr(httpx.Client, "request", _capture([], status=201, body=[])) + row = sc.write_issue_ack("u", "k", "ack") + assert row == {} + + +def test_is_configured_true_when_both_set(monkeypatch): + monkeypatch.setenv("SUPABASE_URL", "http://sb.local") + monkeypatch.setenv("SUPABASE_SERVICE_KEY", "svc-key") + assert sc.is_configured() is True + + +def test_http_error_raises_supabase_error(monkeypatch): + monkeypatch.setenv("SUPABASE_URL", "http://sb.local") + monkeypatch.setenv("SUPABASE_SERVICE_KEY", "svc-key") + + def _fail(self, method, url, **kw): + raise httpx.ConnectError("connection refused") + + monkeypatch.setattr(httpx.Client, "request", _fail) + with pytest.raises(sc.SupabaseError, match="supabase POST issue_acks"): + sc.write_issue_ack("u", "k", "ack")