-
Neda Moeini authoredNeda Moeini authored
processes.py 1.39 KiB
"""Process related endpoints."""
from typing import Any
from uuid import UUID
from fastapi import APIRouter, Depends, HTTPException, status
from orchestrator.db import ProcessStepTable
from orchestrator.schemas.base import OrchestratorBaseModel
from gso.auth.security import opa_security_default
router = APIRouter(prefix="/processes", tags=["Processes"], dependencies=[Depends(opa_security_default)])
class CallBackResultsBaseModel(OrchestratorBaseModel):
"""Base model for callback results."""
callback_results: dict
@router.get(
"/steps/{step_id}/callback-results", status_code=status.HTTP_200_OK, response_model=CallBackResultsBaseModel
)
def callback_results(step_id: UUID) -> dict[str, Any]:
"""Retrieve callback results for a specific process step.
:param step_id: The unique identifier of the process step.
:type step_id: UUID
:return: Dictionary containing callback results.
:rtype: dict[str, Any]
:raises HTTPException: 404 status code if the specified step_id is not found or if the 'callback_result' key
is not present in the state.
"""
step = ProcessStepTable.query.filter(ProcessStepTable.step_id == step_id).first()
if not (step and step.state.get("callback_result", None)):
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Callback result not found.")
return {"callback_results": step.state["callback_result"]}