Skip to content

Wrap a third-party pillar in ~50 LOC

Ophamin's pillars are thin wrappers over mature, battle-tested libraries. The framework's own job is the protocol + the attribution + the test harness; the heavy maths is delegated upstream.

This tutorial walks through wrapping a third-party library as a pillar, using a hypothetical "complexity-from-radon" pillar as the example.

The Pillar protocol

ophamin.protocols.Pillar is a Protocol — anything that quacks like a Pillar can be passed to a scenario. The contract:

class Pillar(Protocol):
    """A statistical / analytical method, delegating to one upstream library."""
    name: str               # short identifier, e.g. "O.spc"
    library: str            # the upstream library used, e.g. "statsmodels"
    library_version: str    # captured at runtime via importlib.metadata

    def run(self, *args, **kwargs) -> Any:
        """Apply the pillar to its inputs; return the result."""

There's no abstract base class; you just implement the three attributes + one method and you're a Pillar. (The register_pillar registry is optional — used when you want CLI discovery.)

Worked example — a complexity pillar over radon

Radon computes cyclomatic complexity + maintainability index. Let's wrap it as an Ophamin pillar.

# src/ophamin/auditing/pillars/radon_pillar.py
from __future__ import annotations

import subprocess
from importlib.metadata import version
from pathlib import Path
from typing import Any

from ophamin.auditing.base import AuditPillar, Finding, FindingSeverity, PillarResult


class RadonComplexityPillar(AuditPillar):
    """Cyclomatic-complexity findings from radon."""

    name = "audit.radon"
    tool_binary = "radon"

    @property
    def library(self) -> str:
        return "radon"

    @property
    def library_version(self) -> str:
        return version("radon")

    def run(self, target_path: str | Path, **_kwargs: Any) -> PillarResult:
        target = str(Path(target_path).resolve())
        if not self.is_available():
            return self.unavailable_result(target)
        cmd = [self.resolved_binary() or self.tool_binary, "cc", "-s", "-j", target]
        completed = subprocess.run(
            cmd, capture_output=True, text=True, check=False
        )
        if completed.returncode != 0:
            return self.error_result(target, completed.stderr or completed.stdout)
        findings = self._parse_radon_json(completed.stdout, target)
        return self.ok_result(
            target,
            findings=findings,
            tool_version=self.library_version,
            wall_time_s=0.0,
        )

    def _parse_radon_json(self, stdout: str, target: str) -> tuple[Finding, ...]:
        import json
        data = json.loads(stdout)
        out: list[Finding] = []
        for path, blocks in data.items():
            for b in blocks:
                if b.get("rank", "") in ("E", "F"):   # rank E/F = high complexity
                    out.append(Finding(
                        pillar_name=self.name,
                        rule_id=f"complexity-{b['rank']}",
                        severity=FindingSeverity.HIGH,
                        message=f"{b['name']} cyclomatic complexity {b['complexity']}",
                        path=path,
                        line=int(b.get("lineno", 0)),
                        column=int(b.get("col_offset", 0)),
                        extra={"complexity": b["complexity"], "rank": b["rank"]},
                    ))
        return tuple(out)

That's ~50 LOC. Now register it:

# src/ophamin/auditing/pillars/__init__.py
from ophamin.auditing.pillars.radon_pillar import RadonComplexityPillar

DEEP_PILLAR_CLASSES = (
    *DEEP_PILLAR_CLASSES,
    RadonComplexityPillar,
)

Test it

# tests/test_audit_radon.py
def test_radon_pillar_skips_when_radon_missing(tmp_path):
    p = RadonComplexityPillar()
    if not p.is_available():
        result = p.run(tmp_path)
        assert result.status == "unavailable"


def test_radon_pillar_finds_high_complexity(tmp_path):
    src = tmp_path / "complex.py"
    src.write_text(
        "def f(x):\n" +
        "\n".join(f"    if x == {i}: return {i}" for i in range(20)) +
        "\n    return -1\n"
    )
    p = RadonComplexityPillar()
    if not p.is_available():
        import pytest; pytest.skip("radon not installed")
    result = p.run(tmp_path)
    assert result.status == "ok"
    assert len(result.findings) >= 1

The "library_version is captured at runtime" rule

Every pillar MUST surface its upstream library + version. Reason: when a proof or audit record is reproduced later, knowing exactly which version of statsmodels / mapie / radon / etc. produced the number is what makes the result auditable.

Don't hardcode the version. Use importlib.metadata.version(...).

The "delegate, don't reinvent" rule

Pillars that reimplement standard maths instead of calling upstream will be rejected at review. The framework's value is the attribution, the test harness, the signed-record output — not re-deriving t-tests in Python.

If the upstream library is missing a feature, file an upstream issue + document the gap. The pillar should still call the library and surface "the feature isn't available in this library version" via a Finding rather than reimplementing it.

Submit it

See Contributing for the PR checklist. For a non-trivial new pillar (new tier? new severity scheme?), open an RFC first.