generated from coulomb/repo-seed
110 lines
3.7 KiB
Python
110 lines
3.7 KiB
Python
from collections.abc import Callable
|
|
from typing import Annotated
|
|
|
|
from fastapi import APIRouter, Depends, Request
|
|
from fastapi.responses import PlainTextResponse, RedirectResponse
|
|
|
|
from core_hub.schemas import CatalogItem, HubCapabilityManifest, HubResource
|
|
from core_hub.security import require_api_token
|
|
from core_hub.seeds import (
|
|
ANNOTATION_CATEGORIES,
|
|
EVENT_TYPES,
|
|
POLICY_SCOPES,
|
|
PUBLIC_HUBS,
|
|
PUBLIC_MANIFESTS,
|
|
WIDGET_TYPES,
|
|
)
|
|
|
|
router = APIRouter(prefix="/api/v2", tags=["api-v2"])
|
|
TokenDependency = Annotated[str, Depends(require_api_token)]
|
|
|
|
|
|
@router.get("/hubs", response_model=list[HubResource])
|
|
def list_hubs() -> list[HubResource]:
|
|
return PUBLIC_HUBS
|
|
|
|
|
|
@router.get("/hub-capability-manifests", response_model=list[HubCapabilityManifest])
|
|
def list_hub_capability_manifests() -> list[HubCapabilityManifest]:
|
|
return PUBLIC_MANIFESTS
|
|
|
|
|
|
@router.get("/widget-types", response_model=list[CatalogItem])
|
|
def list_widget_types() -> list[CatalogItem]:
|
|
return WIDGET_TYPES
|
|
|
|
|
|
@router.get("/event-types", response_model=list[CatalogItem])
|
|
def list_event_types() -> list[CatalogItem]:
|
|
return EVENT_TYPES
|
|
|
|
|
|
@router.get("/annotation-categories", response_model=list[CatalogItem])
|
|
def list_annotation_categories() -> list[CatalogItem]:
|
|
return ANNOTATION_CATEGORIES
|
|
|
|
|
|
@router.get("/policy-scopes", response_model=list[CatalogItem])
|
|
def list_policy_scopes() -> list[CatalogItem]:
|
|
return POLICY_SCOPES
|
|
|
|
|
|
@router.get("/openapi.json", include_in_schema=False)
|
|
def openapi_json(request: Request) -> dict:
|
|
return request.app.openapi()
|
|
|
|
|
|
@router.get("/openapi.yaml", include_in_schema=False)
|
|
def openapi_yaml(request: Request) -> PlainTextResponse:
|
|
info = request.app.openapi().get("info", {})
|
|
body = (
|
|
"openapi: 3.1.0\n"
|
|
"info:\n"
|
|
f" title: {info.get('title', 'Core Hub API')}\n"
|
|
f" version: {info.get('version', '0.1.0')}\n"
|
|
)
|
|
return PlainTextResponse(body, media_type="application/yaml")
|
|
|
|
|
|
@router.get("/docs", include_in_schema=False)
|
|
def docs_redirect() -> RedirectResponse:
|
|
return RedirectResponse(url="/docs")
|
|
|
|
|
|
def protected_empty_collection(
|
|
name: str, method: str
|
|
) -> Callable[[TokenDependency], dict[str, list]]:
|
|
def endpoint(_: TokenDependency) -> dict[str, list]:
|
|
return {name: []}
|
|
|
|
endpoint.__name__ = f"{method.lower()}_{name}_endpoint"
|
|
return endpoint
|
|
|
|
|
|
def register_protected_collection(path: str, name: str, methods: tuple[str, ...]) -> None:
|
|
for method in methods:
|
|
method_lower = method.lower()
|
|
router.add_api_route(
|
|
path,
|
|
protected_empty_collection(name, method),
|
|
methods=[method],
|
|
tags=["api-v2-protected"],
|
|
operation_id=f"{method_lower}_{name}",
|
|
)
|
|
|
|
|
|
register_protected_collection("/api-consumers", "api_consumers", ("GET", "POST"))
|
|
register_protected_collection("/widgets", "widgets", ("GET", "POST"))
|
|
register_protected_collection("/interaction-events", "interaction_events", ("GET", "POST"))
|
|
register_protected_collection("/annotations", "annotations", ("GET", "POST"))
|
|
register_protected_collection("/requirement-candidates", "requirement_candidates", ("GET", "POST"))
|
|
register_protected_collection("/decision-records", "decision_records", ("GET", "POST"))
|
|
register_protected_collection("/deployment-records", "deployment_records", ("GET", "POST"))
|
|
register_protected_collection("/outcome-signals", "outcome_signals", ("GET", "POST"))
|
|
register_protected_collection("/hub-registry", "hub_registry", ("GET",))
|
|
|
|
|
|
@router.post("/token", tags=["api-v2-protected"], operation_id="post_token")
|
|
def issue_token(_: TokenDependency) -> dict[str, str]:
|
|
return {"access_token": "already-authenticated", "token_type": "bearer"}
|