| 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) |
|
|
