Spaces:
Sleeping
Sleeping
File size: 1,783 Bytes
28779c1 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 |
from dataclasses import asdict
import asyncio
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, HttpUrl
from starlette.concurrency import run_in_threadpool
from .engine import assess
from .fetcher import ProfileNotFound, fetch_profile
from .link_resolver import resolve_link
app = FastAPI(title="BMC RiskEngine API", version="1.0.0")
class AssessRequest(BaseModel):
bmc_url: HttpUrl
class ErrorResponse(BaseModel):
detail: str
class AssessResponse(BaseModel):
risk_score: int
risk_level: str
recommended_action: str
evidence: list
limitations: list
async def _analyze(bmc_url: str):
profile = await run_in_threadpool(fetch_profile, bmc_url)
resolved_links = []
# for link in profile.external_links:
# resolved = await run_in_threadpool(resolve_link, link)
# resolved_links.append(resolved)
tasks = [run_in_threadpool(resolve_link, link) for link in profile.external_links]
resolved_links = await asyncio.gather(*tasks)
assessment = assess(profile, resolved_links)
return asdict(assessment)
@app.post(
"/assess", response_model=AssessResponse, responses={404: {"model": ErrorResponse}}
)
async def assess_profile(payload: AssessRequest):
try:
result = await _analyze(str(payload.bmc_url))
except ProfileNotFound as exc:
raise HTTPException(status_code=404, detail="Profile not found") from exc
except Exception as exc:
# TODO: need to log real exceptions
raise HTTPException(status_code=500, detail="Internal server error") from exc
return result
@app.get("/", response_model=dict)
async def root():
return {
"message": "Welcome to the BMC RiskEngine API. Use the /assess endpoint to evaluate a BMC profile."
}
|