Adding a Converter

This skill documents how to add a new output format converter to kgb/io/writers/.

Overview

Converters transform JSON triples into various output formats for use with external tools. The system provides:

• GraphML for graph analysis tools (Gephi, Cytoscape)
• Extensible architecture for custom formats (CSV, RDF, etc.)

Architecture

                        IO Writers Module
    ┌───────────────────────────────────────────────────────────┐
    │                                                           │
    │  io/writers/__init__.py    ← Public exports               │
    │                                                           │
    │  io/writers/graphml.py     ← NetworkX GraphML format      │
    │  ├─ json_to_graphml()        Single file conversion       │
    │  └─ convert_json_directory() Batch conversion             │
    │                                                           │
    │  io/writers/csv.py         ← Your new format              │
    │  ├─ json_to_csv()                                         │
    │  └─ convert_csv_directory()                               │
    │                                                           │
    └───────────────────────────────────────────────────────────┘

Data Flow:
  list[Triple] → Validation → Field Mapping → Format Rendering → File

Key Files:

• kgb/io/writers/graphml.py — Reference implementation (GraphML)
• kgb/io/writers/__init__.py — Public exports
• kgb/io/__init__.py — Top-level IO exports

Dependencies

Field Mapping

Step 1: Understand the Interface

The existing GraphML converter follows this pattern (in kgb/io/writers/graphml.py):

def json_to_graphml(
    triples: list[Triple] | list[dict[str, Any]],
    output_path: Path | str | None = None
) -> nx.DiGraph:
    """Convert triples to a NetworkX DiGraph (optionally saved as GraphML).

    - Validates/converts to Triple objects
    - Normalizes entity names (case-insensitive dedup)
    - Stores relation and inference as edge attributes
    - Uses inference.value (not str(inference)) for clean enum serialization
    """

Key implementation details from the reference:

• Accept both list[Triple] and list[dict] inputs
• Use Triple(**t) to validate dict inputs, skip invalid with warning
• Entity name canonicalization via get_canonical_name() to avoid duplicates
• Preserve inference as .value string ("explicit" / "contextual")

Step 2: Implement Your Converter

Create kgb/io/writers/csv.py:

"""CSV converter for knowledge graph triples."""

from __future__ import annotations
import csv
from pathlib import Path
from typing import Any

from pydantic import ValidationError
from ...domains import Triple


def json_to_csv(
    triples: list[Triple] | list[dict[str, Any]],
    output_path: Path | str,
    *,
    include_metadata: bool = True,
    delimiter: str = ","
) -> Path:
    """Convert triples to CSV edge list format."""
    if not triples:
        raise ValueError("Cannot convert empty triple list")

    output_path = Path(output_path)
    output_path.parent.mkdir(parents=True, exist_ok=True)

    # Validate and convert to Triple objects
    validated: list[Triple] = []
    for t in triples:
        try:
            if isinstance(t, Triple):
                validated.append(t)
            else:
                validated.append(Triple(**t))
        except ValidationError as e:
            print(f"Warning: Skipping invalid triple: {e}")
            continue

    if not validated:
        raise ValueError("No valid triples after validation")

    # Determine columns
    fieldnames = ["head", "relation", "tail"]
    if include_metadata:
        fieldnames.extend(["inference", "justification"])

    # Write CSV
    with open(output_path, "w", newline="", encoding="utf-8") as f:
        writer = csv.DictWriter(f, fieldnames=fieldnames, delimiter=delimiter)
        writer.writeheader()

        for triple in validated:
            row = {
                "head": triple.head,
                "relation": triple.relation,
                "tail": triple.tail,
            }
            if include_metadata:
                row.update({
                    "inference": triple.inference.value,
                    "justification": triple.justification or "",
                })
            writer.writerow(row)

    return output_path


def convert_csv_directory(
    input_dir: Path | str,
    output_dir: Path | str,
    *,
    include_metadata: bool = True
) -> list[Path]:
    """Convert all JSON files to CSV format."""
    import json

    input_dir = Path(input_dir)
    output_dir = Path(output_dir)
    output_dir.mkdir(parents=True, exist_ok=True)

    csv_files = []
    for json_file in input_dir.glob("*.json"):
        try:
            with open(json_file) as f:
                data = json.load(f)

            output_path = output_dir / f"{json_file.stem}.csv"
            json_to_csv(data, output_path, include_metadata=include_metadata)
            print(f"Converted: {json_file.name} -> {output_path.name}")
            csv_files.append(output_path)
        except ValueError as e:
            print(f"Skipped {json_file.name}: {e}")

    return csv_files

Step 3: Register in Module

Update kgb/io/writers/__init__.py:

from .graphml import json_to_graphml, convert_json_directory
from .csv import json_to_csv, convert_csv_directory

__all__ = [
    "json_to_graphml",
    "convert_json_directory",
    "json_to_csv",
    "convert_csv_directory",
]

Update kgb/io/__init__.py to export the new functions:

from .readers import load_records, detect_format, DataLoadError
from .writers import json_to_graphml, convert_json_directory, json_to_csv, convert_csv_directory

__all__ = [
    "load_records",
    "detect_format",
    "DataLoadError",
    "json_to_graphml",
    "convert_json_directory",
    "json_to_csv",
    "convert_csv_directory",
]

Step 4: Add CLI Support

Update the convert command in kgb/__main__.py to support the new format:

@app.command()
def convert(
    input_dir: Path = typer.Option(..., "--input", "-i", exists=True),
    output_dir: Optional[Path] = typer.Option(None, "--output", "-o"),
    format: str = typer.Option("graphml", "--format", "-f"),
):
    """Convert JSON triples to specified format."""
    from .io.writers import convert_json_directory, convert_csv_directory

    out_dir = output_dir or input_dir.parent / format

    if format == "graphml":
        files = convert_json_directory(input_dir, out_dir)
    elif format == "csv":
        files = convert_csv_directory(input_dir, out_dir)
    else:
        console.print(f"[red]Unknown format: {format}[/red]")
        raise typer.Exit(code=1)

    console.print(f"\n[green]Converted {len(files)} files to {format}[/green]")

Step 5: Verify

Check Import

python -c "from kgb.io.writers.csv import json_to_csv; print('OK')"

Unit Tests

def test_json_to_csv_basic(tmp_path):
    from kgb.io.writers.csv import json_to_csv
    from kgb.domains import Triple

    triples = [
        Triple(head="Alice", relation="knows", tail="Bob"),
        Triple(head="Bob", relation="works_at", tail="Acme"),
    ]

    output = tmp_path / "graph.csv"
    result = json_to_csv(triples, output)

    assert result.exists()

    import csv
    with open(result) as f:
        rows = list(csv.DictReader(f))

    assert len(rows) == 2
    assert rows[0]["head"] == "Alice"
    assert rows[0]["inference"] == "explicit"


def test_json_to_csv_empty_list(tmp_path):
    from kgb.io.writers.csv import json_to_csv
    import pytest

    with pytest.raises(ValueError, match="empty"):
        json_to_csv([], tmp_path / "empty.csv")


def test_json_to_csv_from_dicts(tmp_path):
    from kgb.io.writers.csv import json_to_csv
    import csv

    dicts = [{"head": "X", "relation": "r", "tail": "Y", "inference": "explicit"}]
    csv_path = tmp_path / "roundtrip.csv"
    json_to_csv(dicts, csv_path)

    with open(csv_path) as f:
        row = next(csv.DictReader(f))

    assert row["head"] == "X"
    assert row["relation"] == "r"
    assert row["tail"] == "Y"

Key Principles

Error Handling

Files to Create/Modify

Verification Checklist

• Implementation validates Triple inputs
• Uses inference.value for enum serialization
• Tests pass (unit + round-trip)
• Batch function for directory processing
• Registered in kgb/io/writers/__init__.py
• Exported in kgb/io/__init__.py
• CLI format dispatch works (if added)