mirror of
https://github.com/taylanbakircioglu/flowfish.git
synced 2026-09-16 15:45:14 +00:00
taylanbakircioglu
d7ca50b387
Multi-cluster dependency mapping, real-time network monitoring, impact analysis, and CI/CD integration capabilities. Made-with: Cursor
276 lines
10 KiB
Python
276 lines
10 KiB
Python
"""Analysis Management API Endpoints"""
|
|
|
|
from fastapi import APIRouter, HTTPException
|
|
from pydantic import BaseModel
|
|
from typing import List, Optional, Dict, Any
|
|
import grpc
|
|
import sys
|
|
import os
|
|
|
|
# Add proto to path
|
|
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '../..'))
|
|
|
|
from proto import analysis_orchestrator_pb2
|
|
from app.grpc_clients import grpc_clients
|
|
|
|
router = APIRouter(prefix="/analyses", tags=["analyses"])
|
|
|
|
|
|
# Pydantic Models
|
|
class AnalysisCreate(BaseModel):
|
|
cluster_id: int
|
|
name: str
|
|
description: Optional[str] = None
|
|
analysis_type: str # "dependency_mapping", "change_detection", etc.
|
|
parameters: Optional[Dict[str, Any]] = None
|
|
schedule_expression: Optional[str] = None # Cron expression
|
|
created_by: int
|
|
|
|
|
|
class AnalysisResponse(BaseModel):
|
|
id: int
|
|
cluster_id: int
|
|
name: str
|
|
description: str
|
|
analysis_type: str
|
|
status: str
|
|
schedule_expression: str
|
|
is_scheduled: bool
|
|
run_count: int
|
|
|
|
|
|
class AnalysisListResponse(BaseModel):
|
|
analyses: List[AnalysisResponse]
|
|
total_count: int
|
|
|
|
|
|
class AnalysisRunResponse(BaseModel):
|
|
id: int
|
|
analysis_id: int
|
|
status: str
|
|
duration_seconds: int
|
|
result_summary: str
|
|
error_message: str
|
|
|
|
|
|
class AnalysisHistoryResponse(BaseModel):
|
|
runs: List[AnalysisRunResponse]
|
|
|
|
|
|
@router.post("/", response_model=AnalysisResponse)
|
|
async def create_analysis(analysis: AnalysisCreate):
|
|
"""Create a new analysis"""
|
|
try:
|
|
# Map analysis type
|
|
analysis_type_map = {
|
|
"dependency_mapping": analysis_orchestrator_pb2.AnalysisType.DEPENDENCY_MAPPING,
|
|
"change_detection": analysis_orchestrator_pb2.AnalysisType.CHANGE_DETECTION,
|
|
"anomaly_detection": analysis_orchestrator_pb2.AnalysisType.ANOMALY_DETECTION,
|
|
"baseline_creation": analysis_orchestrator_pb2.AnalysisType.BASELINE_CREATION,
|
|
"risk_assessment": analysis_orchestrator_pb2.AnalysisType.RISK_ASSESSMENT,
|
|
}
|
|
|
|
request = analysis_orchestrator_pb2.CreateAnalysisRequest(
|
|
cluster_id=analysis.cluster_id,
|
|
name=analysis.name,
|
|
description=analysis.description or "",
|
|
analysis_type=analysis_type_map.get(
|
|
analysis.analysis_type,
|
|
analysis_orchestrator_pb2.AnalysisType.UNKNOWN_ANALYSIS_TYPE
|
|
),
|
|
parameters=analysis.parameters or {},
|
|
schedule_expression=analysis.schedule_expression or "",
|
|
created_by=analysis.created_by
|
|
)
|
|
|
|
response = grpc_clients.analysis_orchestrator.CreateAnalysis(request)
|
|
|
|
return AnalysisResponse(
|
|
id=response.id,
|
|
cluster_id=response.cluster_id,
|
|
name=response.name,
|
|
description=response.description,
|
|
analysis_type=_map_analysis_type(response.analysis_type),
|
|
status=_map_status(response.status),
|
|
schedule_expression=response.schedule_expression,
|
|
is_scheduled=response.is_scheduled,
|
|
run_count=response.run_count
|
|
)
|
|
|
|
except grpc.RpcError as e:
|
|
raise HTTPException(status_code=500, detail=f"gRPC error: {e.details()}")
|
|
|
|
|
|
@router.get("/", response_model=AnalysisListResponse)
|
|
async def list_analyses(
|
|
cluster_id: Optional[int] = None,
|
|
analysis_type: Optional[str] = None,
|
|
status: Optional[str] = None
|
|
):
|
|
"""List analyses with filters"""
|
|
try:
|
|
# Map filters
|
|
analysis_type_proto = analysis_orchestrator_pb2.AnalysisType.UNKNOWN_ANALYSIS_TYPE
|
|
if analysis_type:
|
|
analysis_type_map = {
|
|
"dependency_mapping": analysis_orchestrator_pb2.AnalysisType.DEPENDENCY_MAPPING,
|
|
"change_detection": analysis_orchestrator_pb2.AnalysisType.CHANGE_DETECTION,
|
|
"anomaly_detection": analysis_orchestrator_pb2.AnalysisType.ANOMALY_DETECTION,
|
|
"baseline_creation": analysis_orchestrator_pb2.AnalysisType.BASELINE_CREATION,
|
|
"risk_assessment": analysis_orchestrator_pb2.AnalysisType.RISK_ASSESSMENT,
|
|
}
|
|
analysis_type_proto = analysis_type_map.get(analysis_type, analysis_orchestrator_pb2.AnalysisType.UNKNOWN_ANALYSIS_TYPE)
|
|
|
|
status_proto = analysis_orchestrator_pb2.AnalysisStatus.UNKNOWN_STATUS
|
|
if status:
|
|
status_map = {
|
|
"pending": analysis_orchestrator_pb2.AnalysisStatus.PENDING,
|
|
"running": analysis_orchestrator_pb2.AnalysisStatus.RUNNING,
|
|
"completed": analysis_orchestrator_pb2.AnalysisStatus.COMPLETED,
|
|
"failed": analysis_orchestrator_pb2.AnalysisStatus.FAILED,
|
|
"cancelled": analysis_orchestrator_pb2.AnalysisStatus.CANCELLED,
|
|
}
|
|
status_proto = status_map.get(status, analysis_orchestrator_pb2.AnalysisStatus.UNKNOWN_STATUS)
|
|
|
|
request = analysis_orchestrator_pb2.ListAnalysesRequest(
|
|
cluster_id=cluster_id or 0,
|
|
analysis_type=analysis_type_proto,
|
|
status=status_proto
|
|
)
|
|
|
|
response = grpc_clients.analysis_orchestrator.ListAnalyses(request)
|
|
|
|
analyses = [
|
|
AnalysisResponse(
|
|
id=a.id,
|
|
cluster_id=a.cluster_id,
|
|
name=a.name,
|
|
description=a.description,
|
|
analysis_type=_map_analysis_type(a.analysis_type),
|
|
status=_map_status(a.status),
|
|
schedule_expression=a.schedule_expression,
|
|
is_scheduled=a.is_scheduled,
|
|
run_count=a.run_count
|
|
)
|
|
for a in response.analyses
|
|
]
|
|
|
|
return AnalysisListResponse(
|
|
analyses=analyses,
|
|
total_count=response.total_count
|
|
)
|
|
|
|
except grpc.RpcError as e:
|
|
raise HTTPException(status_code=500, detail=f"gRPC error: {e.details()}")
|
|
|
|
|
|
@router.get("/{analysis_id}", response_model=AnalysisResponse)
|
|
async def get_analysis(analysis_id: int):
|
|
"""Get analysis by ID"""
|
|
try:
|
|
request = analysis_orchestrator_pb2.GetAnalysisRequest(id=analysis_id)
|
|
response = grpc_clients.analysis_orchestrator.GetAnalysis(request)
|
|
|
|
return AnalysisResponse(
|
|
id=response.id,
|
|
cluster_id=response.cluster_id,
|
|
name=response.name,
|
|
description=response.description,
|
|
analysis_type=_map_analysis_type(response.analysis_type),
|
|
status=_map_status(response.status),
|
|
schedule_expression=response.schedule_expression,
|
|
is_scheduled=response.is_scheduled,
|
|
run_count=response.run_count
|
|
)
|
|
|
|
except grpc.RpcError as e:
|
|
if e.code() == grpc.StatusCode.NOT_FOUND:
|
|
raise HTTPException(status_code=404, detail="Analysis not found")
|
|
raise HTTPException(status_code=500, detail=f"gRPC error: {e.details()}")
|
|
|
|
|
|
@router.delete("/{analysis_id}")
|
|
async def delete_analysis(analysis_id: int):
|
|
"""Delete analysis"""
|
|
try:
|
|
request = analysis_orchestrator_pb2.DeleteAnalysisRequest(id=analysis_id)
|
|
response = grpc_clients.analysis_orchestrator.DeleteAnalysis(request)
|
|
|
|
if not response.success:
|
|
raise HTTPException(status_code=404, detail="Analysis not found")
|
|
|
|
return {"message": "Analysis deleted successfully"}
|
|
|
|
except grpc.RpcError as e:
|
|
raise HTTPException(status_code=500, detail=f"gRPC error: {e.details()}")
|
|
|
|
|
|
@router.post("/{analysis_id}/execute")
|
|
async def execute_analysis(analysis_id: int):
|
|
"""Execute analysis immediately"""
|
|
try:
|
|
request = analysis_orchestrator_pb2.ExecuteAnalysisRequest(id=analysis_id)
|
|
response = grpc_clients.analysis_orchestrator.ExecuteAnalysis(request)
|
|
|
|
return {
|
|
"success": response.success,
|
|
"message": response.message,
|
|
"result_summary": response.result_summary
|
|
}
|
|
|
|
except grpc.RpcError as e:
|
|
raise HTTPException(status_code=500, detail=f"gRPC error: {e.details()}")
|
|
|
|
|
|
@router.get("/{analysis_id}/history", response_model=AnalysisHistoryResponse)
|
|
async def get_analysis_history(analysis_id: int, limit: int = 10):
|
|
"""Get analysis execution history"""
|
|
try:
|
|
request = analysis_orchestrator_pb2.GetAnalysisHistoryRequest(
|
|
analysis_id=analysis_id,
|
|
limit=limit
|
|
)
|
|
response = grpc_clients.analysis_orchestrator.GetAnalysisHistory(request)
|
|
|
|
runs = [
|
|
AnalysisRunResponse(
|
|
id=r.id,
|
|
analysis_id=r.analysis_id,
|
|
status=_map_status(r.status),
|
|
duration_seconds=r.duration_seconds,
|
|
result_summary=r.result_summary,
|
|
error_message=r.error_message
|
|
)
|
|
for r in response.runs
|
|
]
|
|
|
|
return AnalysisHistoryResponse(runs=runs)
|
|
|
|
except grpc.RpcError as e:
|
|
raise HTTPException(status_code=500, detail=f"gRPC error: {e.details()}")
|
|
|
|
|
|
def _map_analysis_type(proto_type) -> str:
|
|
"""Map proto analysis type to string"""
|
|
type_map = {
|
|
analysis_orchestrator_pb2.AnalysisType.DEPENDENCY_MAPPING: "dependency_mapping",
|
|
analysis_orchestrator_pb2.AnalysisType.CHANGE_DETECTION: "change_detection",
|
|
analysis_orchestrator_pb2.AnalysisType.ANOMALY_DETECTION: "anomaly_detection",
|
|
analysis_orchestrator_pb2.AnalysisType.BASELINE_CREATION: "baseline_creation",
|
|
analysis_orchestrator_pb2.AnalysisType.RISK_ASSESSMENT: "risk_assessment",
|
|
}
|
|
return type_map.get(proto_type, "unknown")
|
|
|
|
|
|
def _map_status(proto_status) -> str:
|
|
"""Map proto status to string"""
|
|
status_map = {
|
|
analysis_orchestrator_pb2.AnalysisStatus.PENDING: "pending",
|
|
analysis_orchestrator_pb2.AnalysisStatus.RUNNING: "running",
|
|
analysis_orchestrator_pb2.AnalysisStatus.COMPLETED: "completed",
|
|
analysis_orchestrator_pb2.AnalysisStatus.FAILED: "failed",
|
|
analysis_orchestrator_pb2.AnalysisStatus.CANCELLED: "cancelled",
|
|
}
|
|
return status_map.get(proto_status, "unknown")
|
|
|