File size: 4,869 Bytes
b404de3 | 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 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 | from typing import Any, Dict, List
from cloudenv.server.api.pipeline.schema import (
CreatePipelineRequest,
UpdatePipelineRequest,
)
from cloudenv.server.managers.pipeline_manager import PipelineManager
from cloudenv.server.models.pipeline import Pipeline
from cloudenv.server.providers.aws.adapter import AWSAdapter
from fastapi import APIRouter, Depends, HTTPException, Request
router = APIRouter(prefix="/pipelines", tags=["pipelines"])
def get_pipeline_manager(request: Request) -> PipelineManager:
return request.app.state.pipeline_manager
@router.post("", response_model=Dict[str, str])
async def create_pipeline_from_json(
request: Request,
request_data: CreatePipelineRequest,
pipeline_manager: PipelineManager = Depends(get_pipeline_manager),
) -> Dict[str, str]:
if pipeline_manager.get_pipeline(request_data.pipeline_id):
raise HTTPException(
status_code=400,
detail=f"Pipeline with ID '{request_data.pipeline_id}' already exists",
)
pipeline = Pipeline(pipeline_id=request_data.pipeline_id)
if request_data.broken_pipeline:
pipeline.add_from_json(request_data.broken_pipeline, kind="broken")
if request_data.ideal_pipeline:
pipeline.add_from_json(request_data.ideal_pipeline, kind="ideal")
pipeline_manager.add_pipeline(pipeline)
aws_adapter: "AWSAdapter" = request.app.state.aws_adapter
try:
pipeline.materialize(aws_adapter, episode_id=pipeline.pipeline_id)
except Exception as e:
raise HTTPException(
status_code=500,
detail=f"Failed to materialize pipeline: {str(e)}",
)
return {
"pipeline_id": pipeline.pipeline_id,
"message": "Pipeline created and materialized successfully",
}
@router.get("", response_model=List[Dict])
async def list_pipelines(
pipeline_manager: PipelineManager = Depends(get_pipeline_manager),
) -> List[Dict]:
pipelines_serialized = []
for pipeline in pipeline_manager.pipelines.values():
pipelines_serialized.append(
{
"pipeline_id": pipeline.pipeline_id,
"broken_pipeline": {
k: [n.model_dump() for n in v]
for k, v in pipeline.broken_graph.items()
},
"ideal_pipeline": {
k: [n.model_dump() for n in v]
for k, v in pipeline.ideal_graph.items()
},
}
)
return pipelines_serialized
@router.get("/{pipeline_id}", response_model=Dict[str, Any])
async def get_pipeline(
pipeline_id: str,
pipeline_manager: PipelineManager = Depends(get_pipeline_manager),
) -> Dict[str, Any]:
pipeline = pipeline_manager.get_pipeline(pipeline_id)
if not pipeline:
raise HTTPException(status_code=404, detail="Pipeline not found")
return {
"pipeline_id": pipeline.pipeline_id,
"broken_pipeline": {
k: [n.model_dump() for n in v]
for k, v in pipeline.broken_graph.items()
},
"ideal_pipeline": {
k: [n.model_dump() for n in v]
for k, v in pipeline.ideal_graph.items()
},
}
@router.delete("/{pipeline_id}", response_model=Dict[str, str])
async def delete_pipeline(
pipeline_id: str,
pipeline_manager: PipelineManager = Depends(get_pipeline_manager),
) -> Dict[str, str]:
pipeline = pipeline_manager.get_pipeline(pipeline_id)
if not pipeline:
raise HTTPException(status_code=404, detail="Pipeline not found")
del pipeline_manager.pipelines[pipeline_id]
return {"message": f"Pipeline {pipeline_id} deleted successfully"}
@router.put("/{pipeline_id}", response_model=Dict[str, str])
async def update_pipeline(
pipeline_id: str,
request_data: UpdatePipelineRequest,
pipeline_manager: PipelineManager = Depends(get_pipeline_manager),
) -> Dict[str, str]:
pipeline = pipeline_manager.get_pipeline(pipeline_id)
if not pipeline:
raise HTTPException(status_code=404, detail="Pipeline not found")
pipeline.pipeline_id = request_data.pipeline_id
pipeline_manager.pipelines[request_data.pipeline_id] = pipeline
if pipeline_id != request_data.pipeline_id:
del pipeline_manager.pipelines[pipeline_id]
return {
"pipeline_id": pipeline.pipeline_id,
"message": "Pipeline updated successfully",
}
@router.get("/{pipeline_id}/validate", response_model=Dict[str, Any])
async def validate_pipeline(
pipeline_id: str,
pipeline_manager: PipelineManager = Depends(get_pipeline_manager),
) -> Dict[str, Any]:
pipeline = pipeline_manager.get_pipeline(pipeline_id)
if not pipeline:
raise HTTPException(status_code=404, detail="Pipeline not found")
return pipeline_manager.validate_pipeline(pipeline_id)
|