feat: MVP FastAPI REST API

Endpoints:
- GET /health — health check
- GET /api/v1/regelingen — lijst (met type/status filter)
- GET /api/v1/regelingen/{bwb_id} — metadata
- GET /api/v1/regelingen/{bwb_id}/tekst — volledige Markdown
- GET /api/v1/regelingen/{bwb_id}/artikelen — artikellijst
- GET /api/v1/regelingen/{bwb_id}/artikelen/{nr} — specifiek artikel
- GET /api/v1/regelingen/{bwb_id}/versies — beschikbare toestanden
- GET /api/v1/regelingen/{bwb_id}/diff?van=...&tot=... — versievergelijking
- GET /api/v1/zoeken?q=... — full-text zoeken
- Swagger docs op /api/docs

Getest: zoek "godsdienst" → vindt art. 1, 6, 23 Grondwet.

Sluit #34
This commit is contained in:
Coornhert 2026-03-30 10:27:12 +02:00
parent a7e4a4bc16
commit 21be1367d1
3 changed files with 468 additions and 0 deletions

181
src/wetgit/api/app.py Normal file
View file

@ -0,0 +1,181 @@
"""WetGit REST API — FastAPI applicatie.
Usage:
uvicorn wetgit.api.app:app --host 127.0.0.1 --port 8002
"""
from __future__ import annotations
import os
from pathlib import Path
from fastapi import FastAPI, HTTPException, Query
from fastapi.middleware.cors import CORSMiddleware
from wetgit import __version__
from wetgit.api.data import RegelingStore
from wetgit.api.models import (
ArtikelDetail,
ArtikelItem,
DiffResult,
HealthResponse,
RegelingDetail,
RegelingMeta,
VersieItem,
ZoekResultaat,
)
REPO_PATH = Path(os.environ.get("WETGIT_REPO", "/tmp/wetgit-index-test"))
app = FastAPI(
title="WetGit API",
description="Nederlandse wetgeving als code — REST API",
version=__version__,
docs_url="/api/docs",
redoc_url="/api/redoc",
openapi_url="/api/openapi.json",
)
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_methods=["GET"],
allow_headers=["*"],
)
store = RegelingStore(REPO_PATH)
# --- Health ---
@app.get("/health", response_model=HealthResponse)
def health() -> HealthResponse:
"""Health check."""
return HealthResponse(
status="ok",
version=__version__,
regelingen=len(store.index),
)
# --- Regelingen ---
@app.get("/api/v1/regelingen", response_model=list[RegelingMeta])
def list_regelingen(
type: str | None = Query(None, description="Filter op type (wet, amvb, etc.)"),
status: str | None = Query(None, description="Filter op status (geldend, vervallen)"),
) -> list[dict]:
"""Lijst van alle regelingen."""
regelingen = store.list_regelingen()
if type:
regelingen = [r for r in regelingen if r.get("type") == type]
if status:
regelingen = [r for r in regelingen if r.get("status") == status]
return regelingen
@app.get("/api/v1/regelingen/{bwb_id}", response_model=RegelingMeta)
def get_regeling(bwb_id: str) -> dict:
"""Metadata van een regeling."""
regeling = store.get_regeling(bwb_id)
if not regeling:
raise HTTPException(status_code=404, detail=f"Regeling {bwb_id} niet gevonden")
return regeling
@app.get("/api/v1/regelingen/{bwb_id}/tekst")
def get_tekst(bwb_id: str) -> dict:
"""Volledige tekst van een regeling als Markdown."""
tekst = store.get_tekst(bwb_id)
if tekst is None:
raise HTTPException(status_code=404, detail=f"Regeling {bwb_id} niet gevonden")
return {"bwb_id": bwb_id, "tekst": tekst}
# --- Artikelen ---
@app.get("/api/v1/regelingen/{bwb_id}/artikelen", response_model=list[ArtikelItem])
def list_artikelen(bwb_id: str) -> list[dict]:
"""Lijst van alle artikelen in een regeling."""
artikelen = store.get_artikelen(bwb_id)
if not artikelen:
raise HTTPException(status_code=404, detail=f"Regeling {bwb_id} niet gevonden")
return artikelen
@app.get("/api/v1/regelingen/{bwb_id}/artikelen/{nummer}", response_model=ArtikelDetail)
def get_artikel(bwb_id: str, nummer: str) -> dict:
"""Eén specifiek artikel."""
artikel = store.get_artikel(bwb_id, nummer)
if not artikel:
raise HTTPException(
status_code=404,
detail=f"Artikel {nummer} niet gevonden in {bwb_id}",
)
return artikel
# --- Versies & Diff ---
@app.get("/api/v1/regelingen/{bwb_id}/versies", response_model=list[VersieItem])
def list_versies(bwb_id: str) -> list[dict]:
"""Beschikbare versies (toestanden) van een regeling."""
versies = store.get_versies(bwb_id)
if not versies:
raise HTTPException(status_code=404, detail=f"Regeling {bwb_id} niet gevonden")
return versies
@app.get("/api/v1/regelingen/{bwb_id}/diff", response_model=DiffResult)
def get_diff(
bwb_id: str,
van: str = Query(..., description="Startdatum (YYYY-MM-DD)"),
tot: str = Query(..., description="Einddatum (YYYY-MM-DD)"),
) -> dict:
"""Vergelijk twee versies van een regeling."""
result = store.get_diff(bwb_id, van, tot)
if not result:
raise HTTPException(
status_code=404,
detail=f"Geen diff beschikbaar voor {bwb_id} tussen {van} en {tot}",
)
return result
# --- Zoeken ---
@app.get("/api/v1/zoeken", response_model=list[ZoekResultaat])
def zoeken(
q: str = Query(..., min_length=2, description="Zoekterm"),
type: str | None = Query(None, description="Filter op type"),
limit: int = Query(20, ge=1, le=100, description="Max resultaten"),
) -> list[dict]:
"""Doorzoek alle wetgeving (full-text)."""
import re
resultaten: list[dict] = []
for regeling in store.list_regelingen():
if type and regeling.get("type") != type:
continue
tekst = store.get_tekst(regeling["bwb_id"])
if tekst is None or q.lower() not in tekst.lower():
continue
# Zoek in welk artikel de match zit
current_artikel = ""
for line in tekst.split("\n"):
if line.startswith("### Artikel"):
current_artikel = line.replace("### ", "")
if q.lower() in line.lower() and current_artikel:
resultaten.append({
"bwb_id": regeling["bwb_id"],
"titel": regeling.get("titel", ""),
"artikel": current_artikel,
"context": line.strip()[:200],
})
if len(resultaten) >= limit:
return resultaten
return resultaten

209
src/wetgit/api/data.py Normal file
View file

@ -0,0 +1,209 @@
"""Data-laag — leest regelingen uit de Markdown bestanden."""
from __future__ import annotations
import json
import re
import subprocess
from functools import lru_cache
from pathlib import Path
import frontmatter
class RegelingStore:
"""Leest en cached regelingen uit de wetgit/rijk repo."""
def __init__(self, repo_path: Path) -> None:
self.repo_path = repo_path
self._index: list[dict] | None = None
@property
def index(self) -> list[dict]:
if self._index is None:
self._index = self._load_index()
return self._index
def reload(self) -> None:
"""Herlaad de index."""
self._index = None
def _load_index(self) -> list[dict]:
"""Laad index.json of scan de repo."""
index_path = self.repo_path / "index.json"
if index_path.exists():
data = json.loads(index_path.read_text(encoding="utf-8"))
return data.get("regelingen", [])
# Fallback: scan de repo
from wetgit.pipeline.indexer import generate_index
return generate_index(self.repo_path)
def list_regelingen(self) -> list[dict]:
"""Lijst van alle regelingen."""
return self.index
def get_regeling(self, bwb_id: str) -> dict | None:
"""Haal metadata op voor één regeling."""
for r in self.index:
if r["bwb_id"] == bwb_id:
return r
return None
def get_tekst(self, bwb_id: str) -> str | None:
"""Haal de volledige Markdown tekst op."""
md_path = self._find_md(bwb_id)
if md_path is None:
return None
return md_path.read_text(encoding="utf-8")
def get_artikelen(self, bwb_id: str) -> list[dict]:
"""Haal alle artikelen op als lijst."""
tekst = self.get_tekst(bwb_id)
if tekst is None:
return []
return self._extract_artikelen(tekst)
def get_artikel(self, bwb_id: str, nummer: str) -> dict | None:
"""Haal één specifiek artikel op."""
tekst = self.get_tekst(bwb_id)
if tekst is None:
return None
pattern = rf"(### Artikel {re.escape(nummer)}\b.*?)(?=\n### Artikel |\n## |\Z)"
match = re.search(pattern, tekst, re.DOTALL)
if not match:
return None
art_text = match.group(1).strip()
titel = None
lines = art_text.split("\n")
for line in lines[1:]:
if line.startswith("*") and line.endswith("*") and not line.startswith("**"):
titel = line.strip("*").strip()
break
return {
"nummer": nummer,
"titel": titel,
"tekst": art_text,
"bwb_id": bwb_id,
}
def get_versies(self, bwb_id: str) -> list[dict]:
"""Haal beschikbare versies op via git log."""
md_path = self._find_md(bwb_id)
if md_path is None:
return []
rel_path = md_path.relative_to(self.repo_path)
try:
result = subprocess.run(
["git", "log", "--format=%ai %s", "--follow", "--", str(rel_path)],
cwd=self.repo_path, capture_output=True, text=True, check=True,
)
except subprocess.CalledProcessError:
return []
versies = []
for line in result.stdout.strip().split("\n"):
if not line:
continue
datum = line.split(" ")[0]
versies.append({"datum": datum})
return versies
def get_diff(self, bwb_id: str, van: str, tot: str) -> dict | None:
"""Vergelijk twee versies."""
md_path = self._find_md(bwb_id)
if md_path is None:
return None
rel_path = md_path.relative_to(self.repo_path)
# Zoek commits bij de datums
try:
log_result = subprocess.run(
["git", "log", "--format=%H %ai", "--follow", "--", str(rel_path)],
cwd=self.repo_path, capture_output=True, text=True, check=True,
)
except subprocess.CalledProcessError:
return None
commits: list[tuple[str, str]] = []
for line in log_result.stdout.strip().split("\n"):
if line:
parts = line.split(" ")
commits.append((parts[0], parts[1]))
commit_van = self._find_commit_for_date(commits, van)
commit_tot = self._find_commit_for_date(commits, tot)
if not commit_van or not commit_tot or commit_van == commit_tot:
return None
try:
diff_result = subprocess.run(
["git", "diff", commit_van, commit_tot, "--", str(rel_path)],
cwd=self.repo_path, capture_output=True, text=True, check=True,
)
except subprocess.CalledProcessError:
return None
diff_text = diff_result.stdout
added = sum(1 for l in diff_text.split("\n") if l.startswith("+") and not l.startswith("+++"))
removed = sum(1 for l in diff_text.split("\n") if l.startswith("-") and not l.startswith("---"))
return {
"bwb_id": bwb_id,
"van": van,
"tot": tot,
"diff": diff_text,
"regels_toegevoegd": added,
"regels_verwijderd": removed,
}
def _find_md(self, bwb_id: str) -> Path | None:
"""Zoek het Markdown bestand voor een BWB-ID."""
for md in self.repo_path.rglob("README.md"):
if bwb_id in str(md):
return md
return None
def _find_commit_for_date(
self, commits: list[tuple[str, str]], target: str,
) -> str | None:
"""Vind de commit die het dichtst bij de datum ligt."""
for commit_hash, date_str in commits:
if date_str <= target:
return commit_hash
return commits[-1][0] if commits else None
def _extract_artikelen(self, tekst: str) -> list[dict]:
"""Extraheer artikelen uit Markdown tekst."""
artikelen = []
pattern = r"### Artikel (\S+)(.*?)(?=\n### Artikel |\n## |\Z)"
for match in re.finditer(pattern, tekst, re.DOTALL):
nummer = match.group(1)
body = match.group(2).strip()
titel = None
preview = ""
for line in body.split("\n"):
line = line.strip()
if not line:
continue
if line.startswith("*") and line.endswith("*") and not line.startswith("**"):
titel = line.strip("*").strip()
else:
preview = line[:200]
break
artikelen.append({
"nummer": nummer,
"titel": titel,
"preview": preview,
})
return artikelen

78
src/wetgit/api/models.py Normal file
View file

@ -0,0 +1,78 @@
"""Pydantic models voor de REST API."""
from __future__ import annotations
from pydantic import BaseModel, Field
class RegelingMeta(BaseModel):
"""Metadata van een regeling."""
bwb_id: str
titel: str
citeertitel: str | None = None
type: str
status: str
datum_inwerkingtreding: str | None = None
bron: str
pad: str
artikelen: int
structuur_elementen: int
class RegelingDetail(RegelingMeta):
"""Regeling met volledige tekst."""
tekst: str
class ArtikelItem(BaseModel):
"""Artikel in een lijst."""
nummer: str
titel: str | None = None
preview: str = Field(description="Eerste 200 tekens van het artikel")
class ArtikelDetail(BaseModel):
"""Volledig artikel."""
nummer: str
titel: str | None = None
tekst: str
bwb_id: str
class VersieItem(BaseModel):
"""Beschikbare versie (toestand)."""
datum: str
geldig_tot: str | None = None
class DiffResult(BaseModel):
"""Resultaat van een versievergelijking."""
bwb_id: str
van: str
tot: str
diff: str
regels_toegevoegd: int
regels_verwijderd: int
class ZoekResultaat(BaseModel):
"""Eén zoekresultaat."""
bwb_id: str
titel: str
artikel: str
context: str
class HealthResponse(BaseModel):
"""Health check response."""
status: str = "ok"
version: str
regelingen: int