Core API

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.

class core.base_probe.ProbeResult(probe: str, source_type: str, source_id: str, timestamp: str = '', records: list[dict] = <factory>, record_count: int = 0, duration_ms: int = 0, error: str | None = None)

Standard envelope for all probe outputs.

probe: str

source_type: str

source_id: str

timestamp: str = ''

records: list[dict]

record_count: int = 0

duration_ms: int = 0

error: str | None = None

to_dict() → dict

to_json(indent: int = 2) → str

class core.base_probe.BaseProbe

All probes inherit from this. Implement probe_name() and execute().

Optional: set record_model to a Pydantic BaseModel subclass to declare the expected record schema. Used for documentation and optional validation.

record_model: type | None = None

abstractmethod static probe_name() → str

Fully qualified probe name, e.g., ‘oracle.metadata.tables’.

abstractmethod static description() → str

One-line description of what this probe extracts.

abstractmethod execute(connection, schema: str, **kwargs) → list[dict]

Run the probe against a connection. Return list of record dicts.

run(connection, schema: str, source_id: str = '', **kwargs) → ProbeResult

Execute the probe with timing, error handling, logging, and envelope wrapping.

write(result: ProbeResult, output_dir: Path) → Path

Write probe result to local JSON file.

Probe registry — discovers and runs probes by pattern matching.

Supports two discovery mechanisms: 1. Built-in: auto-discover probes under probes/ package 2. Plugins: external packages register via entry_points group ‘metapod.probes’

To create a plugin:

# In your package’s pyproject.toml: [project.entry-points.”metapod.probes”] my_source = “my_package.probes”

core.registry.register(cls: type[BaseProbe]) → type[BaseProbe]

Decorator to register a probe class.

core.registry.get_probe(name: str) → type[BaseProbe] | None

core.registry.list_probes(pattern: str = '*') → list[str]

List registered probe names matching a glob pattern.

core.registry.get_probes(pattern: str = '*') → list[type[BaseProbe]]

Get probe classes matching a glob pattern.

core.registry.discover_probes()

Auto-discover all probes: built-in + plugins.

Configuration loader — YAML + environment variables.

class core.config.SourceConfig(*, type: str, host: str = 'localhost', port: int = 1521, service_name: str = '', sid: str = '', username: str = '', password: SecretStr = SecretStr(''), schema_name: str = '', workspace_url: str = '', token: SecretStr = SecretStr(''), catalog: str = '', timeout: int = 120)

type: str

host: str

port: int

service_name: str

sid: str

username: str

password: SecretStr

schema_name: str

workspace_url: str

token: SecretStr

catalog: str

timeout: int

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class core.config.TargetConfig(*, url: str = '', api_key: SecretStr = SecretStr(''), project: str = 'default')

url: str

api_key: SecretStr

project: str

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class core.config.ProbeConfig(*, source: SourceConfig, target: TargetConfig = TargetConfig(url='', project='default'), output_dir: str = './output', probes: list[str] = ['*'], schedule: str = '')

source: SourceConfig

target: TargetConfig

output_dir: str

probes: list[str]

schedule: str

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

core.config.load_config(path: str | Path) → ProbeConfig

Load config from YAML file with environment variable substitution.

Supports ${VAR_NAME} syntax in string values.

Transport layer — push probe results to target platform API with retry logic.

core.transport.push_probes(output_dir: Path, target: TargetConfig, scan: bool = True) → dict

Push all probe results to target platform’s /api/probes/ingest endpoint.

This is the recommended push method. The platform converts probe JSON into DDL/PL/SQL files, stores them in Metastore, and optionally scans.

core.transport.push_directory(output_dir: Path, target: TargetConfig) → list[dict]

Push all JSON files from output directory to target platform Metastore.

Probe diff — compare current vs previous extraction results.

Detects: added, removed, and changed records between two probe runs. Useful for change detection and drift monitoring.

core.diff.diff_probes(current_dir: Path, previous_dir: Path) → list[dict]

Compare all probe JSON files between two output directories.

Returns a list of diff records: {probe, added, removed, changed, unchanged}

Simple cron scheduler for recurring probe execution.

Uses a minimal cron parser — no APScheduler dependency needed. Runs as a blocking loop, suitable for Docker or systemd.

Usage:

metapod schedule –cron “0 6 * * “ “oracle.metadata.”

core.scheduler.matches_cron(cron_expr: str, dt: datetime) → bool

Check if a datetime matches a cron expression (minute hour day month weekday).

core.scheduler.run_schedule(cron_expr: str, callback, check_interval: int = 30)

Run callback when cron expression matches. Blocks forever.

Parameters:

• cron_expr – Cron expression (minute hour day month weekday)

• callback – Function to call (no args)

• check_interval – Seconds between checks (default 30)