Part 3: OOP, Dataclasses and Protocols

Introduction

ansible-inspec has two major subsystems that need to share a common interface: the Ansible adapter (which builds playbooks and inventories) and the InSpec adapter (which parses Ruby-based profiles). Both must be swappable for testing purposes — I can inject a fake adapter in tests without touching real inventory files or Ruby parsers.

That design lives on three Python OOP features: @dataclass, Protocol, and standard class inheritance. This part covers each.

---

Classes and __init__

The fundamental building block:

class ProfileConverter:
    """Converts an InSpec profile directory into an Ansible collection."""

    # Class variable — shared across all instances
    supported_resources: list[str] = [
        "file", "service", "package",
        "security_policy", "registry_key", "audit_policy",
    ]

    def __init__(self, profile_path: str, namespace: str = "local") -> None:
        self.profile_path = profile_path
        self.namespace = namespace
        self._controls: list[dict] = []   # private by convention

    def load(self) -> None:
        """Parse controls from the profile directory."""
        import os
        for root, _, files in os.walk(self.profile_path):
            for f in files:
                if f.endswith(".rb"):
                    self._parse_file(os.path.join(root, f))

    def _parse_file(self, path: str) -> None:
        """Internal — parse a single .rb control file."""
        with open(path) as f:
            content = f.read()
        # Simplified: real impl does Ruby AST parsing
        self._controls.append({"path": path, "raw": content})

    def convert(self) -> dict:
        """Return the Ansible collection structure as a dict."""
        if not self._controls:
            self.load()
        return {
            "namespace": self.namespace,
            "controls": self._controls,
        }

---

Inheritance

The converter has multiple output targets. Inheritance lets us share the parsing logic:

class BaseConverter:
    def __init__(self, profile_path: str) -> None:
        self.profile_path = profile_path
        self._controls: list[dict] = []

    def load(self) -> None:
        import os
        for root, _, files in os.walk(self.profile_path):
            for f in files:
                if f.endswith(".rb"):
                    self._parse_file(os.path.join(root, f))

    def _parse_file(self, path: str) -> None:
        with open(path) as f:
            self._controls.append({"path": path, "raw": f.read()})

    def convert(self) -> dict:
        raise NotImplementedError("Subclasses must implement convert()")


class AnsibleCollectionConverter(BaseConverter):
    def __init__(self, profile_path: str, namespace: str, collection_name: str) -> None:
        super().__init__(profile_path)
        self.namespace = namespace
        self.collection_name = collection_name

    def convert(self) -> dict:
        if not self._controls:
            self.load()
        return {
            "type": "ansible_collection",
            "namespace": self.namespace,
            "name": self.collection_name,
            "roles": self._build_roles(),
        }

    def _build_roles(self) -> list[dict]:
        return [{"name": "compliance_check", "tasks": self._controls}]


class JSONExporter(BaseConverter):
    def convert(self) -> dict:
        if not self._controls:
            self.load()
        return {"type": "json_export", "controls": self._controls}

super() — calling parent methods

class TimedConverter(AnsibleCollectionConverter):
    def convert(self) -> dict:
        from datetime import datetime
        start = datetime.now()
        result = super().convert()           # call parent's convert()
        elapsed = (datetime.now() - start).total_seconds()
        result["conversion_time_seconds"] = elapsed
        return result

---

@dataclass

@dataclass auto-generates __init__, __repr__, __eq__ from field annotations. I use dataclasses for lightweight value objects that don't need Pydantic's validation overhead:

from dataclasses import dataclass, field
from datetime import datetime

@dataclass
class ControlResult:
    control_id: str
    title: str
    status: str
    message: str | None = None

@dataclass
class HostReport:
    hostname: str
    controls: list[ControlResult] = field(default_factory=list)
    checked_at: datetime = field(default_factory=datetime.now)

    def passed(self) -> list[ControlResult]:
        return [c for c in self.controls if c.status == "passed"]

    def failed(self) -> list[ControlResult]:
        return [c for c in self.controls if c.status == "failed"]

    def summary(self) -> dict[str, int]:
        return {
            "passed": len(self.passed()),
            "failed": len(self.failed()),
            "total": len(self.controls),
        }


# Usage
report = HostReport(hostname="web-01")
report.controls.append(ControlResult("sshd-01", "PermitRootLogin", "passed"))
report.controls.append(ControlResult("pkg-01", "telnet not installed", "failed",
                                      message="telnetd 0.17 is installed"))
print(report.summary())  # {'passed': 1, 'failed': 1, 'total': 2}

@dataclass(frozen=True) — immutable data

from dataclasses import dataclass

@dataclass(frozen=True)
class ConnectionKey:
    """Immutable key used for connection pooling."""
    host: str
    port: int
    user: str

    def uri(self) -> str:
        return f"ssh://{self.user}@{self.host}:{self.port}"

key = ConnectionKey("web-01", 22, "deploy")
# key.host = "web-02"  # raises FrozenInstanceError

@dataclass vs Pydantic

	@dataclass	Pydantic BaseModel
Validation	None (you add it manually)	Automatic on assignment
JSON serialise	Manual	.model_dump_json()
Performance	Very fast	Slightly more overhead (still fast with v2)
Use for	Internal value objects	API request/response, config

---

Protocol — Structural Subtyping

Protocol is Python's duck-typing formalised. Instead of requiring isinstance checks, you define the interface:

from typing import Protocol, runtime_checkable

# Any class with these methods satisfies the protocol
@runtime_checkable
class InventoryLoader(Protocol):
    def load(self, path: str) -> dict[str, list[str]]: ...
    def validate(self) -> bool: ...


class AnsibleInventoryLoader:
    """Reads Ansible YAML inventory."""

    def load(self, path: str) -> dict[str, list[str]]:
        import yaml
        with open(path) as f:
            raw = yaml.safe_load(f)
        hosts = raw.get("all", {}).get("hosts", {})
        return {"hosts": list(hosts.keys())}

    def validate(self) -> bool:
        return True  # simplified


class StaticInventoryLoader:
    """For tests — returns a fixed inventory without reading files."""

    def __init__(self, hosts: list[str]) -> None:
        self._hosts = hosts

    def load(self, path: str) -> dict[str, list[str]]:
        return {"hosts": self._hosts}

    def validate(self) -> bool:
        return len(self._hosts) > 0


# A function that accepts any InventoryLoader
def run_compliance(
    loader: InventoryLoader,
    inventory_path: str,
    profile: str,
) -> dict:
    if not loader.validate():
        raise ValueError("Invalid inventory loader")
    inventory = loader.load(inventory_path)
    return {"profile": profile, "hosts": inventory["hosts"]}


# Both work — Protocol is satisfied structurally, no inheritance needed
real_loader = AnsibleInventoryLoader()
test_loader = StaticInventoryLoader(["web-01", "web-02"])

print(isinstance(real_loader, InventoryLoader))   # True (runtime_checkable)
print(isinstance(test_loader, InventoryLoader))   # True

This is exactly how ansible-inspec separates the real Ansible adapter from test doubles.

---

Abstract Base Classes (ABC)

When you want to enforce that subclasses implement certain methods at class definition time (not at call time), use ABC:

from abc import ABC, abstractmethod

class BaseReporter(ABC):
    """All reporters must implement these methods."""

    @abstractmethod
    def render(self, results: list[HostReport]) -> str:
        """Convert results to a string output."""
        ...

    @abstractmethod
    def save(self, output: str, path: str) -> None:
        """Save the rendered output to a file."""
        ...

    # Concrete method shared by all reporters
    def report(self, results: list[HostReport], path: str) -> None:
        output = self.render(results)
        self.save(output, path)


class JSONReporter(BaseReporter):
    def render(self, results: list[HostReport]) -> str:
        import json
        data = [
            {"host": r.hostname, "summary": r.summary()}
            for r in results
        ]
        return json.dumps(data, indent=2)

    def save(self, output: str, path: str) -> None:
        with open(path, "w") as f:
            f.write(output)


class CLIReporter(BaseReporter):
    def render(self, results: list[HostReport]) -> str:
        lines = []
        for r in results:
            s = r.summary()
            lines.append(
                f"{r.hostname}: {s['passed']} passed, {s['failed']} failed"
            )
        return "\n".join(lines)

    def save(self, output: str, path: str) -> None:
        print(output)   # CLI reporter just prints

ABC vs Protocol

	ABC	Protocol
Enforcement	At class definition (TypeError on instantiation)	At type-check time (mypy/pyright)
Inheritance required	Yes	No — structural match is enough
Runtime check	isinstance(x, ABC)	isinstance(x, Protocol) with @runtime_checkable
Use for	Shared base with concrete helpers	Interface contract without coupling

---

Class Methods and Static Methods

from dataclasses import dataclass
from datetime import datetime
import json

@dataclass
class JobTemplate:
    name: str
    profile: str
    timeout: int = 300

    @classmethod
    def from_dict(cls, data: dict) -> "JobTemplate":
        """Alternative constructor from a raw dict."""
        return cls(
            name=data["name"],
            profile=data["profile"],
            timeout=data.get("timeout", 300),
        )

    @classmethod
    def from_json(cls, json_str: str) -> "JobTemplate":
        return cls.from_dict(json.loads(json_str))

    @staticmethod
    def is_valid_profile_name(name: str) -> bool:
        """Utility that doesn't need self or cls."""
        import re
        return bool(re.match(r"^[\w\-/]+$", name))


t = JobTemplate.from_dict({"name": "ssh-baseline", "profile": "dev-sec/ssh-baseline"})
print(JobTemplate.is_valid_profile_name("dev-sec/linux-baseline"))  # True
print(JobTemplate.is_valid_profile_name("bad name!"))               # False

---

Properties

from dataclasses import dataclass, field
from datetime import datetime

@dataclass
class JobExecution:
    job_id: str
    started_at: datetime | None = None
    finished_at: datetime | None = None
    _status: str = field(default="pending", repr=False)

    @property
    def status(self) -> str:
        return self._status

    @status.setter
    def status(self, value: str) -> None:
        allowed = {"pending", "running", "success", "failed"}
        if value not in allowed:
            raise ValueError(f"Invalid status: {value}. Must be one of {allowed}")
        self._status = value

    @property
    def duration(self) -> float | None:
        if self.started_at and self.finished_at:
            return (self.finished_at - self.started_at).total_seconds()
        return None


job = JobExecution(job_id="abc-123")
job.status = "running"
job.started_at = datetime.now()
# job.status = "invalid"  # raises ValueError

---

Putting It Together

Here is a simplified slice of how ansible-inspec's adapter layer is structured:

from typing import Protocol
from dataclasses import dataclass, field

# ---- Domain types ----

@dataclass
class Profile:
    name: str
    path: str
    controls: list[str] = field(default_factory=list)

@dataclass
class CheckResult:
    host: str
    passed: int
    failed: int
    skipped: int

# ---- Protocols (interfaces) ----

class ProfileLoader(Protocol):
    def load_profile(self, path: str) -> Profile: ...

class ComplianceRunner(Protocol):
    def run(self, profile: Profile, hosts: list[str]) -> list[CheckResult]: ...

# ---- Concrete implementations ----

class LocalProfileLoader:
    def load_profile(self, path: str) -> Profile:
        import os
        controls = [
            f for f in os.listdir(path) if f.endswith(".rb")
        ]
        return Profile(name=os.path.basename(path), path=path, controls=controls)

class AnsibleRunner:
    def run(self, profile: Profile, hosts: list[str]) -> list[CheckResult]:
        # In real code this shells out to ansible-playbook
        return [
            CheckResult(host=h, passed=len(profile.controls),
                        failed=0, skipped=0)
            for h in hosts
        ]

# ---- Orchestrator that depends only on protocols ----

class ComplianceOrchestrator:
    def __init__(self, loader: ProfileLoader, runner: ComplianceRunner) -> None:
        self._loader = loader
        self._runner = runner

    def execute(self, profile_path: str, hosts: list[str]) -> list[CheckResult]:
        profile = self._loader.load_profile(profile_path)
        return self._runner.run(profile, hosts)


# Production
orchestrator = ComplianceOrchestrator(
    loader=LocalProfileLoader(),
    runner=AnsibleRunner(),
)

# Test — inject fakes, no file system needed
class FakeLoader:
    def load_profile(self, path: str) -> Profile:
        return Profile(name="test", path=path, controls=["ctrl-01"])

class FakeRunner:
    def run(self, profile: Profile, hosts: list[str]) -> list[CheckResult]:
        return [CheckResult(h, 1, 0, 0) for h in hosts]

test_orchestrator = ComplianceOrchestrator(FakeLoader(), FakeRunner())

---

Summary

Concept	When to use
Plain class	When you need methods and shared state
@dataclass	Lightweight value objects; auto-generates boilerplate
@dataclass(frozen=True)	Immutable value objects, safe dict keys
ABC	Shared base with enforced abstract interface
Protocol	Interface-only contract without inheritance coupling
@classmethod	Alternative constructors
@staticmethod	Utility functions tied to the class logically
@property	Computed attributes with validation on set

---

What's Next

Part 4 covers asyncio, async/await, and how ansible-inspec's FastAPI server handles concurrent compliance job execution without blocking the event loop.