Phase 7: API Integration

Objective

Expose the association graph as HTTP endpoints in domus-api. After this phase, curl localhost:8080/associations/CISSP | jq returns the same data as the CLI, but accessible from any HTTP client.

This follows the same patterns you have already seen in domus-api: a service layer for business logic, a routes module for HTTP handling, and lifespan loading for startup initialization.

Concept	Plain English
FastAPI route	A function decorated with `@router.get("/path")` that handles HTTP requests. When a request hits that path, FastAPI calls your function and returns its result as JSON. You have already seen this in domus-api’s existing routes.
Path parameter	A variable segment in a URL: `/associations/{key}` means `key` is extracted from the URL. `GET /associations/CISSP` passes `key="CISSP"` to your function. This is the `{key}` in the route, like `$1` in a positional parameter.
`APIRouter`	A way to organize routes into modules. Each module creates a `router`, and the main app includes it with `app.include_router(router)`. This is modular design — like sourcing separate files in a bash script.
Lifespan	FastAPI’s startup/shutdown hook. Code in the lifespan function runs once when the server starts (load the graph) and can clean up when it stops. domus-api already uses this pattern for `DocumentCache`.
`app.state`	A place to store shared objects (like the loaded graph) that all routes can access. `request.app.state.graph` retrieves it from any route handler.
`Depends()`	Dependency injection — FastAPI calls a function to provide a value to your route handler. You declare what you need; FastAPI figures out how to get it. This keeps route functions clean and testable.
Service layer	A module that holds business logic separate from HTTP concerns. The route handles the request/response; the service handles the data operations. This separation means you can use the same logic from the CLI, the API, and tests.

A function decorated with @router.get("/path") that handles HTTP requests. When a request hits that path, FastAPI calls your function and returns its result as JSON. You have already seen this in domus-api’s existing routes.

Path parameter

A variable segment in a URL: /associations/{key} means key is extracted from the URL. GET /associations/CISSP passes key="CISSP" to your function. This is the {key} in the route, like $1 in a positional parameter.

APIRouter

A way to organize routes into modules. Each module creates a router, and the main app includes it with app.include_router(router). This is modular design — like sourcing separate files in a bash script.

Lifespan

FastAPI’s startup/shutdown hook. Code in the lifespan function runs once when the server starts (load the graph) and can clean up when it stops. domus-api already uses this pattern for DocumentCache.

app.state

A place to store shared objects (like the loaded graph) that all routes can access. request.app.state.graph retrieves it from any route handler.

Depends()

Dependency injection — FastAPI calls a function to provide a value to your route handler. You declare what you need; FastAPI figures out how to get it. This keeps route functions clean and testable.

Service layer

A module that holds business logic separate from HTTP concerns. The route handles the request/response; the service handles the data operations. This separation means you can use the same logic from the CLI, the API, and tests.

Steps

1. Create the service module

Create src/association_engine/services/associations.py (and services/__init__.py):

"""Association graph service — business logic layer."""

from pathlib import Path

from association_engine.graph import AssociationGraph

# Default data directory
DATA_DIR = Path("data")


class AssociationService:
    """Wraps AssociationGraph with application-level operations."""

    def __init__(self, data_dir: Path = DATA_DIR) -> None:
        self._graph = AssociationGraph.load_directory(data_dir)
        self._data_dir = data_dir

    def query(self, key: str) -> dict[str, list[str]]:
        """Forward lookup."""
        return self._graph.query(key)

    def reverse_query(self, key: str) -> dict[str, list[str]]:
        """Reverse lookup."""
        return self._graph.reverse_query(key)

    def keys(self) -> list[str]:
        """All known entities."""
        return self._graph.keys()

    def relations(self) -> list[str]:
        """All relation types."""
        return self._graph.relations()

    def associate(
        self, source: str, relation: str, target: str, file: str = "api.yml"
    ) -> None:
        """Add an association and persist it."""
        self._graph.associate(source, relation, target)
        self._graph.save(self._data_dir / file)

The service is a thin wrapper. It owns the graph instance, handles persistence decisions (which file to save to), and provides a clean interface for both the API and CLI.

2. Create the routes module

Create src/association_engine/routes/associations.py (and routes/__init__.py):

"""Association graph HTTP endpoints."""

from fastapi import APIRouter, HTTPException, Request
from pydantic import BaseModel

router = APIRouter(prefix="/associations", tags=["associations"])


class AssociationCreate(BaseModel):
    """Request body for creating an association."""

    source: str
    relation: str
    target: str


def _get_service(request: Request):
    """Retrieve the AssociationService from app state."""
    return request.app.state.association_service


@router.get("/{key}")
async def get_associations(key: str, request: Request):
    """Return all forward associations for an entity.

    Example: GET /associations/CISSP
    """
    service = _get_service(request)
    result = service.query(key)

    if not result:
        raise HTTPException(status_code=404, detail=f"No associations for '{key}'")

    return {"entity": key, "associations": result}


@router.get("/{key}/reverse")
async def get_reverse_associations(key: str, request: Request):
    """Return all reverse associations for an entity.

    Example: GET /associations/access-control/reverse
    """
    service = _get_service(request)
    result = service.reverse_query(key)

    if not result:
        raise HTTPException(status_code=404, detail=f"No reverse associations for '{key}'")

    return {"entity": key, "reverse_associations": result}


@router.get("/")
async def list_entities(request: Request):
    """Return all known entities in the graph."""
    service = _get_service(request)
    return {"entities": service.keys(), "count": len(service.keys())}


@router.post("/", status_code=201)
async def create_association(body: AssociationCreate, request: Request):
    """Add a new association.

    Example: POST /associations
    Body: {"source": "CISSP", "relation": "covers", "target": "risk-management"}
    """
    service = _get_service(request)
    service.associate(body.source, body.relation, body.target)

    return {
        "status": "created",
        "association": {
            "source": body.source,
            "relation": body.relation,
            "target": body.target,
        },
    }

Key design decisions:

_get_service() retrieves the service from app.state — this is the same pattern domus-api uses for DocumentCache.
HTTPException(status_code=404) returns a proper 404 when an entity is not found.
BaseModel (Pydantic) validates the POST request body — if source, relation, or target is missing, FastAPI returns a 422 with a clear error message.
Routes are async because FastAPI is an async framework. The graph operations are synchronous, but the async declaration lets FastAPI handle concurrent requests efficiently.

3. Register the routes in the app

In your domus-api main application file, add:

from association_engine.routes.associations import router as associations_router
from association_engine.services.associations import AssociationService

# In the lifespan function (alongside DocumentCache loading):
app.state.association_service = AssociationService(data_dir=Path("data"))

# After app creation:
app.include_router(associations_router)

This follows the existing pattern — domus-api already has a lifespan function and router registration. The association service loads at startup, just like the document cache.

4. Test the endpoints

Start the server:

uv run uvicorn association_engine.main:app --port 8080

Then in another terminal:

# Forward query
curl -s localhost:8080/associations/CISSP | jq

# Reverse query
curl -s localhost:8080/associations/access-control/reverse | jq

# List all entities
curl -s localhost:8080/associations/ | jq '.entities'

# Add a new association
curl -s -X POST localhost:8080/associations/ \
  -H "Content-Type: application/json" \
  -d '{"source": "CCNP", "relation": "covers", "target": "BGP"}' | jq

# Verify it was added
curl -s localhost:8080/associations/CCNP | jq

5. Write API tests

Create tests/test_api.py:

"""Tests for the association API endpoints."""

from pathlib import Path

import pytest
from fastapi.testclient import TestClient

from association_engine.services.associations import AssociationService


@pytest.fixture
def client(tmp_path: Path) -> TestClient:
    """Create a test client with a temporary data directory."""
    from fastapi import FastAPI
    from association_engine.routes.associations import router

    # Create test data
    from association_engine.graph import AssociationGraph

    g = AssociationGraph()
    g.associate("CISSP", "covers", "access-control")
    g.associate("CISSP", "covers", "cryptography")
    g.save(tmp_path / "test.yml")

    app = FastAPI()
    app.state.association_service = AssociationService(data_dir=tmp_path)
    app.include_router(router)

    return TestClient(app)


class TestGetAssociations:
    def test_returns_associations(self, client: TestClient) -> None:
        resp = client.get("/associations/CISSP")
        assert resp.status_code == 200
        data = resp.json()
        assert "access-control" in data["associations"]["covers"]

    def test_unknown_returns_404(self, client: TestClient) -> None:
        resp = client.get("/associations/nonexistent")
        assert resp.status_code == 404


class TestReverseAssociations:
    def test_returns_reverse(self, client: TestClient) -> None:
        resp = client.get("/associations/access-control/reverse")
        assert resp.status_code == 200
        data = resp.json()
        assert "CISSP" in data["reverse_associations"]["covered-by"]


class TestCreateAssociation:
    def test_creates_association(self, client: TestClient) -> None:
        resp = client.post(
            "/associations/",
            json={"source": "Python", "relation": "uses", "target": "typer"},
        )
        assert resp.status_code == 201

        # Verify it's queryable
        resp = client.get("/associations/Python")
        assert resp.status_code == 200

TestClient is FastAPI’s test utility — it simulates HTTP requests without starting a real server. The test fixture creates a temporary data directory, populates it, and builds a mini FastAPI app.

6. Run all tests

uv run pytest tests/ -v --tb=short

Checklist

services/associations.py created with AssociationService
routes/associations.py created with four endpoints
Routes registered in the main app
GET /associations/{key} returns forward associations
GET /associations/{key}/reverse returns reverse associations
POST /associations creates new associations
GET /associations/ lists all entities
API tests pass
All previous tests still pass
curl verification works

Verification

# Start server in background
uv run uvicorn association_engine.main:app --port 8080 &
SERVER_PID=$!

# Test
curl -sf localhost:8080/associations/CISSP | jq -e '.associations.covers | length > 0'
RESULT=$?

# Cleanup
kill $SERVER_PID

# Report
[ $RESULT -eq 0 ] && echo "API: ok" || echo "API: FAIL"

# Tests
uv run pytest tests/ -v --tb=short

The association graph is now accessible three ways: as a Python library, as a CLI tool, and as an HTTP API. Phase 8 populates it with real data from your domus-captures knowledge base.