Skip to content
Snippets Groups Projects

Feature/nat 468 refactor auth

Merged Mohammad Torkashvand requested to merge feature/NAT-468-refactor-auth into develop
5 files
+ 3
94
Compare changes
  • Side-by-side
  • Inline
Files
5
+ 0
40
"""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 orchestrator.security import authorize
router = APIRouter(prefix="/processes", tags=["Processes"], dependencies=[Depends(authorize)])
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"]}
Loading