generated from coulomb/repo-seed
291 lines
12 KiB
Python
291 lines
12 KiB
Python
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
|
|
import httpx
|
|
from fastapi import APIRouter, Body, Depends, HTTPException, Query
|
|
from sqlalchemy import func, select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from api.database import get_session
|
|
from api.models.fabric_graph import FabricGraphEdge, FabricGraphImport, FabricGraphNode
|
|
from api.schemas.fabric_graph import (
|
|
FabricGraphEdgeRead,
|
|
FabricGraphImportRead,
|
|
FabricGraphIngestResult,
|
|
FabricGraphNodeRead,
|
|
FabricGraphPullRequest,
|
|
FabricGraphSummary,
|
|
)
|
|
from api.services.fabric_graph import (
|
|
FabricGraphValidationError,
|
|
ingest_fabric_graph_export,
|
|
record_fabric_graph_error,
|
|
split_graph_ingest_body,
|
|
)
|
|
|
|
router = APIRouter(prefix="/fabric", tags=["fabric"])
|
|
|
|
|
|
@router.post("/graph-exports", response_model=FabricGraphIngestResult)
|
|
async def ingest_graph_export(
|
|
body: dict[str, Any] = Body(...),
|
|
source_repo_slug: str | None = Query(None),
|
|
source_url: str | None = Query(None),
|
|
session: AsyncSession = Depends(get_session),
|
|
) -> FabricGraphIngestResult:
|
|
graph, metadata = split_graph_ingest_body(body)
|
|
effective_source_repo_slug = (
|
|
source_repo_slug
|
|
or metadata.get("source_repo_slug")
|
|
or metadata.get("repo_slug")
|
|
or "railiance-fabric"
|
|
)
|
|
effective_source_url = source_url or metadata.get("source_url")
|
|
requested_by = str(metadata.get("requested_by") or "api")
|
|
try:
|
|
import_run, created, idempotent = await ingest_fabric_graph_export(
|
|
session,
|
|
graph,
|
|
source_repo_slug=str(effective_source_repo_slug),
|
|
source_url=str(effective_source_url) if effective_source_url else None,
|
|
requested_by=requested_by,
|
|
)
|
|
except FabricGraphValidationError as exc:
|
|
raise HTTPException(status_code=422, detail=exc.detail) from exc
|
|
return FabricGraphIngestResult(
|
|
import_run=FabricGraphImportRead.model_validate(import_run),
|
|
created=created,
|
|
idempotent=idempotent,
|
|
node_count=import_run.node_count,
|
|
edge_count=import_run.edge_count,
|
|
)
|
|
|
|
|
|
@router.post("/graph-exports/pull", response_model=FabricGraphIngestResult)
|
|
async def pull_graph_export(
|
|
body: FabricGraphPullRequest | None = None,
|
|
session: AsyncSession = Depends(get_session),
|
|
) -> FabricGraphIngestResult:
|
|
request = body or FabricGraphPullRequest()
|
|
try:
|
|
async with httpx.AsyncClient(timeout=15.0) as client:
|
|
response = await client.get(request.source_url)
|
|
response.raise_for_status()
|
|
payload = response.json()
|
|
except Exception as exc:
|
|
await record_fabric_graph_error(
|
|
session,
|
|
"Fabric graph export pull failed.",
|
|
source_repo_slug=request.source_repo_slug,
|
|
source_url=request.source_url,
|
|
error=str(exc),
|
|
requested_by=request.requested_by,
|
|
)
|
|
raise HTTPException(status_code=502, detail=f"Fabric graph export pull failed: {exc}") from exc
|
|
|
|
try:
|
|
import_run, created, idempotent = await ingest_fabric_graph_export(
|
|
session,
|
|
payload,
|
|
source_repo_slug=request.source_repo_slug,
|
|
source_url=request.source_url,
|
|
requested_by=request.requested_by,
|
|
)
|
|
except FabricGraphValidationError as exc:
|
|
raise HTTPException(status_code=422, detail=exc.detail) from exc
|
|
return FabricGraphIngestResult(
|
|
import_run=FabricGraphImportRead.model_validate(import_run),
|
|
created=created,
|
|
idempotent=idempotent,
|
|
node_count=import_run.node_count,
|
|
edge_count=import_run.edge_count,
|
|
)
|
|
|
|
|
|
@router.get("/graph-exports", response_model=list[FabricGraphImportRead])
|
|
async def list_graph_imports(
|
|
source_repo_slug: str | None = None,
|
|
validation_status: str | None = None,
|
|
limit: int = Query(50, ge=1, le=500),
|
|
session: AsyncSession = Depends(get_session),
|
|
) -> list[FabricGraphImportRead]:
|
|
query = select(FabricGraphImport)
|
|
if source_repo_slug:
|
|
query = query.where(FabricGraphImport.source_repo_slug == source_repo_slug)
|
|
if validation_status:
|
|
query = query.where(FabricGraphImport.validation_status == validation_status)
|
|
query = query.order_by(FabricGraphImport.created_at.desc()).limit(limit)
|
|
result = await session.execute(query)
|
|
return [FabricGraphImportRead.model_validate(row) for row in result.scalars().all()]
|
|
|
|
|
|
@router.get("/graph-exports/latest", response_model=FabricGraphImportRead)
|
|
async def latest_graph_import(
|
|
source_repo_slug: str = "railiance-fabric",
|
|
session: AsyncSession = Depends(get_session),
|
|
) -> FabricGraphImportRead:
|
|
import_run = await _latest_valid_import(session, source_repo_slug)
|
|
if import_run is None:
|
|
raise HTTPException(status_code=404, detail=f"No valid Fabric graph import for '{source_repo_slug}'")
|
|
return FabricGraphImportRead.model_validate(import_run)
|
|
|
|
|
|
@router.get("/graph/nodes", response_model=list[FabricGraphNodeRead])
|
|
async def list_graph_nodes(
|
|
source_repo_slug: str = "railiance-fabric",
|
|
domain: str | None = None,
|
|
repo: str | None = None,
|
|
canonical_category: str | None = None,
|
|
evidence_state: str | None = None,
|
|
mapping_fit: str | None = None,
|
|
kind: str | None = None,
|
|
limit: int = Query(100, ge=1, le=1000),
|
|
session: AsyncSession = Depends(get_session),
|
|
) -> list[FabricGraphNodeRead]:
|
|
import_run = await _latest_valid_import_or_404(session, source_repo_slug)
|
|
query = select(FabricGraphNode).where(FabricGraphNode.import_id == import_run.id)
|
|
if domain:
|
|
query = query.where(FabricGraphNode.domain_slug == domain)
|
|
if repo:
|
|
query = query.where(FabricGraphNode.repo_slug == repo)
|
|
if canonical_category:
|
|
query = query.where(FabricGraphNode.canon_category == canonical_category)
|
|
if evidence_state:
|
|
query = query.where(FabricGraphNode.evidence_state == evidence_state)
|
|
if mapping_fit:
|
|
query = query.where(FabricGraphNode.mapping_fit == mapping_fit)
|
|
if kind:
|
|
query = query.where(FabricGraphNode.kind == kind)
|
|
query = query.order_by(FabricGraphNode.graph_id).limit(limit)
|
|
result = await session.execute(query)
|
|
return [FabricGraphNodeRead.model_validate(row) for row in result.scalars().all()]
|
|
|
|
|
|
@router.get("/graph/edges", response_model=list[FabricGraphEdgeRead])
|
|
async def list_graph_edges(
|
|
source_repo_slug: str = "railiance-fabric",
|
|
canonical_relationship: str | None = None,
|
|
edge_type: str | None = None,
|
|
evidence_state: str | None = None,
|
|
mapping_fit: str | None = None,
|
|
display_only: bool | None = None,
|
|
from_graph_id: str | None = None,
|
|
to_graph_id: str | None = None,
|
|
limit: int = Query(100, ge=1, le=1000),
|
|
session: AsyncSession = Depends(get_session),
|
|
) -> list[FabricGraphEdgeRead]:
|
|
import_run = await _latest_valid_import_or_404(session, source_repo_slug)
|
|
query = select(FabricGraphEdge).where(FabricGraphEdge.import_id == import_run.id)
|
|
if canonical_relationship:
|
|
query = query.where(FabricGraphEdge.canonical_type == canonical_relationship)
|
|
if edge_type:
|
|
query = query.where(FabricGraphEdge.edge_type == edge_type)
|
|
if evidence_state:
|
|
query = query.where(FabricGraphEdge.evidence_state == evidence_state)
|
|
if mapping_fit:
|
|
query = query.where(FabricGraphEdge.mapping_fit == mapping_fit)
|
|
if display_only is not None:
|
|
query = query.where(FabricGraphEdge.display_only == display_only)
|
|
if from_graph_id:
|
|
query = query.where(FabricGraphEdge.from_graph_id == from_graph_id)
|
|
if to_graph_id:
|
|
query = query.where(FabricGraphEdge.to_graph_id == to_graph_id)
|
|
query = query.order_by(FabricGraphEdge.from_graph_id, FabricGraphEdge.to_graph_id, FabricGraphEdge.edge_type).limit(limit)
|
|
result = await session.execute(query)
|
|
return [FabricGraphEdgeRead.model_validate(row) for row in result.scalars().all()]
|
|
|
|
|
|
@router.get("/graph/summary", response_model=FabricGraphSummary)
|
|
async def graph_summary(
|
|
source_repo_slug: str = "railiance-fabric",
|
|
session: AsyncSession = Depends(get_session),
|
|
) -> FabricGraphSummary:
|
|
import_run = await _latest_valid_import(session, source_repo_slug)
|
|
if import_run is None:
|
|
return FabricGraphSummary(
|
|
source_repo_slug=source_repo_slug,
|
|
latest_import=None,
|
|
node_count=0,
|
|
edge_count=0,
|
|
nodes_by_domain={},
|
|
nodes_by_repo={},
|
|
nodes_by_canon_category={},
|
|
edges_by_canonical_type={},
|
|
nodes_by_evidence_state={},
|
|
edges_by_evidence_state={},
|
|
nodes_by_mapping_fit={},
|
|
edges_by_mapping_fit={},
|
|
example_nodes=[],
|
|
example_edges=[],
|
|
)
|
|
|
|
return FabricGraphSummary(
|
|
source_repo_slug=source_repo_slug,
|
|
latest_import=FabricGraphImportRead.model_validate(import_run),
|
|
node_count=import_run.node_count,
|
|
edge_count=import_run.edge_count,
|
|
nodes_by_domain=await _counts(session, FabricGraphNode.domain_slug, import_run.id),
|
|
nodes_by_repo=await _counts(session, FabricGraphNode.repo_slug, import_run.id),
|
|
nodes_by_canon_category=await _counts(session, FabricGraphNode.canon_category, import_run.id),
|
|
edges_by_canonical_type=await _counts(session, FabricGraphEdge.canonical_type, import_run.id),
|
|
nodes_by_evidence_state=await _counts(session, FabricGraphNode.evidence_state, import_run.id),
|
|
edges_by_evidence_state=await _counts(session, FabricGraphEdge.evidence_state, import_run.id),
|
|
nodes_by_mapping_fit=await _counts(session, FabricGraphNode.mapping_fit, import_run.id),
|
|
edges_by_mapping_fit=await _counts(session, FabricGraphEdge.mapping_fit, import_run.id),
|
|
example_nodes=await _example_nodes(session, import_run.id),
|
|
example_edges=await _example_edges(session, import_run.id),
|
|
)
|
|
|
|
|
|
async def _latest_valid_import(session: AsyncSession, source_repo_slug: str) -> FabricGraphImport | None:
|
|
result = await session.execute(
|
|
select(FabricGraphImport)
|
|
.where(
|
|
FabricGraphImport.source_repo_slug == source_repo_slug,
|
|
FabricGraphImport.validation_status == "valid",
|
|
FabricGraphImport.is_latest.is_(True),
|
|
)
|
|
.order_by(FabricGraphImport.created_at.desc())
|
|
.limit(1)
|
|
)
|
|
return result.scalar_one_or_none()
|
|
|
|
|
|
async def _latest_valid_import_or_404(session: AsyncSession, source_repo_slug: str) -> FabricGraphImport:
|
|
import_run = await _latest_valid_import(session, source_repo_slug)
|
|
if import_run is None:
|
|
raise HTTPException(status_code=404, detail=f"No valid Fabric graph import for '{source_repo_slug}'")
|
|
return import_run
|
|
|
|
|
|
async def _counts(session: AsyncSession, column: Any, import_id: Any) -> dict[str, int]:
|
|
table = column.class_
|
|
result = await session.execute(
|
|
select(column, func.count())
|
|
.select_from(table)
|
|
.where(table.import_id == import_id, column.is_not(None))
|
|
.group_by(column)
|
|
)
|
|
return {str(key): int(count) for key, count in result.all() if key is not None}
|
|
|
|
|
|
async def _example_nodes(session: AsyncSession, import_id: Any) -> list[FabricGraphNodeRead]:
|
|
result = await session.execute(
|
|
select(FabricGraphNode)
|
|
.where(FabricGraphNode.import_id == import_id)
|
|
.order_by(FabricGraphNode.graph_id)
|
|
.limit(5)
|
|
)
|
|
return [FabricGraphNodeRead.model_validate(row) for row in result.scalars().all()]
|
|
|
|
|
|
async def _example_edges(session: AsyncSession, import_id: Any) -> list[FabricGraphEdgeRead]:
|
|
result = await session.execute(
|
|
select(FabricGraphEdge)
|
|
.where(FabricGraphEdge.import_id == import_id)
|
|
.order_by(FabricGraphEdge.from_graph_id, FabricGraphEdge.to_graph_id, FabricGraphEdge.edge_type)
|
|
.limit(5)
|
|
)
|
|
return [FabricGraphEdgeRead.model_validate(row) for row in result.scalars().all()]
|