"""Base probe interface — every probe implements this contract.
A probe is a single unit of work: connect → extract → format → write.
Probes are stateless, idempotent, and produce a JSON envelope.
"""
from __future__ import annotations
import json
import time
from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from datetime import datetime, timezone
from pathlib import Path
[docs]
@dataclass
class ProbeResult:
"""Standard envelope for all probe outputs."""
probe: str # e.g., "oracle.metadata.tables"
source_type: str # e.g., "oracle"
source_id: str # e.g., "prod-db:1521/RISK_MGMT"
timestamp: str = ""
records: list[dict] = field(default_factory=list)
record_count: int = 0
duration_ms: int = 0
error: str | None = None
[docs]
def to_dict(self) -> dict:
return {
"probe": self.probe,
"source": {"type": self.source_type, "id": self.source_id},
"timestamp": self.timestamp,
"records": self.records,
"record_count": self.record_count,
"duration_ms": self.duration_ms,
"error": self.error,
}
[docs]
def to_json(self, indent: int = 2) -> str:
return json.dumps(self.to_dict(), indent=indent, ensure_ascii=False, default=str)
[docs]
class BaseProbe(ABC):
"""All probes inherit from this. Implement probe_name() and execute()."""
[docs]
@staticmethod
@abstractmethod
def probe_name() -> str:
"""Fully qualified probe name, e.g., 'oracle.metadata.tables'."""
...
[docs]
@staticmethod
@abstractmethod
def description() -> str:
"""One-line description of what this probe extracts."""
...
[docs]
@abstractmethod
def execute(self, connection, schema: str, **kwargs) -> list[dict]:
"""Run the probe against a connection. Return list of record dicts."""
...
[docs]
def run(self, connection, schema: str, source_id: str = "", **kwargs) -> ProbeResult:
"""Execute the probe with timing, error handling, and envelope wrapping."""
result = ProbeResult(
probe=self.probe_name(),
source_type=self.probe_name().split(".")[0],
source_id=source_id or schema,
timestamp=datetime.now(timezone.utc).isoformat(),
)
start = time.monotonic()
try:
records = self.execute(connection, schema, **kwargs)
result.records = records
result.record_count = len(records)
except Exception as e:
result.error = str(e)
result.duration_ms = int((time.monotonic() - start) * 1000)
return result
[docs]
def write(self, result: ProbeResult, output_dir: Path) -> Path:
"""Write probe result to local JSON file."""
# File path: output_dir/{source_type}/{schema}/{probe_leaf}.json
parts = self.probe_name().split(".")
category = parts[1] if len(parts) > 1 else "unknown"
leaf = parts[-1]
out_path = output_dir / parts[0] / category / f"{leaf}.json"
out_path.parent.mkdir(parents=True, exist_ok=True)
out_path.write_text(result.to_json(), encoding="utf-8")
return out_path