Hands-on · ~25 min

API Exposure

FastAPI endpoints over the extracted data: list agreements, fetch one with obligations, search by supplier. The shape a customer's BI / app would call.

What we're building

A small REST API:

GET /agreements — list, with filter by supplier or term_end.
GET /agreements/{id} — full record, including linked documents.
GET /suppliers — list.
GET /renewals/upcoming — agreements with upcoming renewals.

No auth, no pagination beyond limit. Production would add both; for the build-along we keep it minimal.

FastAPI app

src/ci/api.py

from __future__ import annotations
from datetime import date, timedelta
from typing import Annotated
from fastapi import FastAPI, HTTPException, Query
from sqlalchemy import select
from sqlalchemy.orm import Session

from ci.db import SessionLocal
from ci.models import Agreement, Supplier, Document, DocumentAgreementLink, Extraction
from ci.api_schemas import AgreementOut, AgreementDetail, DocumentOut, SupplierOut, RenewalOut


app = FastAPI(title="Contract Intelligence API")


def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()


@app.get("/healthz")
def healthz():
    return {"ok": True}

Response schemas

src/ci/api_schemas.py

from __future__ import annotations
from datetime import date
from pydantic import BaseModel


class SupplierOut(BaseModel):
    supplier_id: str
    canonical_name: str
    country: str | None = None


class AgreementOut(BaseModel):
    agreement_id: str
    supplier: SupplierOut
    agreement_type: str
    effective_date: date | None
    current_term_end: date | None
    status: str


class DocumentOut(BaseModel):
    document_id: str
    doc_class: str
    page_count: int
    role: str  # master | amendment | sow | exhibit


class AgreementDetail(AgreementOut):
    documents: list[DocumentOut] = []
    extracted_fields: dict | None = None


class RenewalOut(AgreementOut):
    days_until_renewal: int

Endpoints

src/ci/api.py — endpoints

from fastapi import Depends


@app.get("/suppliers", response_model=list[SupplierOut])
def list_suppliers(db: Session = Depends(get_db),
                   limit: int = Query(100, le=1000)):
    rows = db.execute(select(Supplier).limit(limit)).scalars().all()
    return [SupplierOut(supplier_id=r.supplier_id,
                        canonical_name=r.canonical_name,
                        country=r.country) for r in rows]


@app.get("/agreements", response_model=list[AgreementOut])
def list_agreements(
    db: Session = Depends(get_db),
    supplier_id: str | None = Query(None),
    status: str | None = Query(None),
    limit: int = Query(100, le=1000),
):
    q = select(Agreement, Supplier).join(Supplier, Supplier.supplier_id == Agreement.supplier_id)
    if supplier_id:
        q = q.where(Agreement.supplier_id == supplier_id)
    if status:
        q = q.where(Agreement.status == status)
    q = q.limit(limit)
    rows = db.execute(q).all()
    return [
        AgreementOut(
            agreement_id=a.agreement_id,
            supplier=SupplierOut(supplier_id=s.supplier_id, canonical_name=s.canonical_name, country=s.country),
            agreement_type=a.agreement_type,
            effective_date=a.effective_date,
            current_term_end=a.current_term_end,
            status=a.status,
        )
        for a, s in rows
    ]


@app.get("/agreements/{agreement_id}", response_model=AgreementDetail)
def get_agreement(agreement_id: str, db: Session = Depends(get_db)):
    row = db.execute(
        select(Agreement, Supplier)
        .join(Supplier, Supplier.supplier_id == Agreement.supplier_id)
        .where(Agreement.agreement_id == agreement_id)
    ).first()
    if not row:
        raise HTTPException(404, "Agreement not found")
    a, s = row

    docs = db.execute(
        select(Document, DocumentAgreementLink.role)
        .join(DocumentAgreementLink, DocumentAgreementLink.document_id == Document.document_id)
        .where(DocumentAgreementLink.agreement_id == agreement_id)
    ).all()
    doc_outs = [DocumentOut(document_id=d.document_id, doc_class=d.doc_class,
                            page_count=d.page_count, role=role)
                for d, role in docs]

    # Pull the master document's extraction
    master_doc = next((d for d, r in docs if r == "master"), None)
    extracted_fields = None
    if master_doc:
        ext = db.execute(
            select(Extraction).where(Extraction.document_id == master_doc.document_id)
            .order_by(Extraction.extracted_at.desc())
            .limit(1)
        ).scalar_one_or_none()
        if ext:
            extracted_fields = ext.fields

    return AgreementDetail(
        agreement_id=a.agreement_id,
        supplier=SupplierOut(supplier_id=s.supplier_id, canonical_name=s.canonical_name, country=s.country),
        agreement_type=a.agreement_type,
        effective_date=a.effective_date,
        current_term_end=a.current_term_end,
        status=a.status,
        documents=doc_outs,
        extracted_fields=extracted_fields,
    )


@app.get("/renewals/upcoming", response_model=list[RenewalOut])
def upcoming_renewals(
    db: Session = Depends(get_db),
    days: int = Query(90, le=365),
):
    cutoff = date.today() + timedelta(days=days)
    rows = db.execute(
        select(Agreement, Supplier)
        .join(Supplier, Supplier.supplier_id == Agreement.supplier_id)
        .where(Agreement.status == "active")
        .where(Agreement.current_term_end <= cutoff)
        .where(Agreement.current_term_end >= date.today())
        .order_by(Agreement.current_term_end)
    ).all()
    return [
        RenewalOut(
            agreement_id=a.agreement_id,
            supplier=SupplierOut(supplier_id=s.supplier_id, canonical_name=s.canonical_name, country=s.country),
            agreement_type=a.agreement_type,
            effective_date=a.effective_date,
            current_term_end=a.current_term_end,
            status=a.status,
            days_until_renewal=(a.current_term_end - date.today()).days,
        )
        for a, s in rows
    ]

Run it

uvicorn ci.api:app --reload --port 8000

Open http://localhost:8000/docs — FastAPI's auto-generated Swagger UI. You can hit every endpoint from the browser.

Verify

curl http://localhost:8000/suppliers | jq
curl http://localhost:8000/agreements | jq
curl 'http://localhost:8000/renewals/upcoming?days=180' | jq

# Pick an agreement_id from the list:
curl http://localhost:8000/agreements/agr_demo_abc123 | jq

You should see:

Suppliers populated with the canonical names extracted in chapter 04.
Agreements with their effective_date and current_term_end.
Per-agreement detail including linked documents and extracted fields with confidence + provenance.
Renewals filtered to the window you specify.

Test

tests/test_api.py

from fastapi.testclient import TestClient
from ci.api import app


def test_healthz():
    client = TestClient(app)
    resp = client.get("/healthz")
    assert resp.status_code == 200
    assert resp.json() == {"ok": True}


def test_list_agreements_empty():
    client = TestClient(app)
    resp = client.get("/agreements?limit=10")
    assert resp.status_code == 200
    assert isinstance(resp.json(), list)

Run pytest tests/test_api.py -v. With a populated database, you can extend tests to verify specific records.

The pipeline is now complete end-to-end: PDFs → documents → extractions → agreements → API.

✓ Check yourself

Answer from memory before opening each one.

Why does the extraction lookup in get_agreement sort by extracted_at and take one row, instead of selecting every extraction for the document?

Show answer
The unique key is document plus model version, so a document gathers one extraction per model. Selecting by document alone returns several rows, and scalar_one_or_none raises when it gets more than one, which would turn into a 500. Newest first with a limit of one returns the latest result. See Endpoints.
get_db never calls commit, while db_session in chapter 02 does. Is that a bug?

Show answer
No. Every endpoint here only reads, so there is nothing to commit, and the dependency closes the session after the response. The first write endpoint you add needs a commit, or it will silently discard its changes when the session closes. See FastAPI app.
A 12-month agreement with auto-renewal ended 20 days ago and no longer appears in /renewals/upcoming, though the contract is still in force. Why, and what would you change?

Show answer
The filter keeps only term ends from today forward, and current_term_end is computed once from the effective date and first term. Nothing rolls it forward after renewal, so every auto-renewing contract drops out after its first term. Compute the next renewal date from the renewal terms, or advance the stored date. See Endpoints.
None of the queries filter by customer_id. What happens when a second customer's data is loaded, and where would you enforce the boundary?

Show answer
Every endpoint returns every customer's rows. A WHERE clause that each query must remember is easy to forget, so enforce it in the database with row-level security keyed to the authenticated tenant. Add authentication in the API first, since the tenant comes from it. See What we're building.
Run curl 'http://localhost:8000/agreements?limit=5000'. What status do you get, and why is that better than returning 1,000 rows?

Show answer
A 422 validation error, because the parameter declares an upper bound of 1,000 and FastAPI rejects values above it. Silent clamping would hide that the client asked for something the service will not give. See Endpoints.

Takeaway

Each request gets its own database session through a dependency that closes it afterwards, and a read-only API never needs a commit.
Response models declare exactly what leaves the service, so a column such as tax_id stays private until someone adds it on purpose.
A lookup that can match several rows, such as one extraction per model, must sort and limit or it fails when data accumulates.
A stored term end that is never rolled forward makes auto-renewing contracts vanish from the renewal list after one term.
Without a tenant filter or row-level security, every endpoint serves every customer's data.
Out-of-range parameters should be rejected loudly, so a client learns the limit instead of getting a truncated answer.

The pipeline now runs from PDF to API, and the last chapter maps where to take it from here. → Extensions & Next Steps