""" Camunda BPMN Engine API Proxy Routes API requests from /api/bpmn/* to the Camunda 7 REST API """ import os import httpx from fastapi import APIRouter, Request, HTTPException, Response, UploadFile, File, Form from fastapi.responses import JSONResponse from typing import Optional # Camunda REST API URL CAMUNDA_URL = os.getenv("CAMUNDA_URL", "http://camunda:8080/engine-rest") ENVIRONMENT = os.getenv("ENVIRONMENT", "development") router = APIRouter(prefix="/bpmn", tags=["bpmn"]) async def proxy_request(request: Request, path: str, timeout: float = 30.0) -> Response: """Forward a request to the Camunda REST API""" url = f"{CAMUNDA_URL}{path}" # Forward headers headers = {} if "authorization" in request.headers: headers["Authorization"] = request.headers["authorization"] if "content-type" in request.headers: headers["Content-Type"] = request.headers["content-type"] # Get request body for POST/PUT/PATCH/DELETE body = None if request.method in ("POST", "PUT", "PATCH", "DELETE"): body = await request.body() async with httpx.AsyncClient(timeout=timeout) as client: try: response = await client.request( method=request.method, url=url, headers=headers, content=body, params=request.query_params ) return Response( content=response.content, status_code=response.status_code, headers=dict(response.headers), media_type=response.headers.get("content-type", "application/json") ) except httpx.ConnectError: raise HTTPException( status_code=503, detail="Camunda BPMN engine unavailable" ) except httpx.TimeoutException: raise HTTPException( status_code=504, detail="Camunda BPMN engine timeout" ) # ============================================ # Health & Engine Info # ============================================ @router.get("/health") async def health(): """Health check for Camunda connection""" async with httpx.AsyncClient(timeout=5.0) as client: try: response = await client.get(f"{CAMUNDA_URL}/engine") if response.status_code == 200: engines = response.json() return { "camunda": "healthy", "connected": True, "engines": engines } return {"camunda": "unhealthy", "connected": False} except Exception as e: return {"camunda": "unhealthy", "connected": False, "error": str(e)} @router.get("/engine") async def get_engines(request: Request): """Get list of process engines""" return await proxy_request(request, "/engine") # ============================================ # Process Definitions # ============================================ @router.get("/process-definition") async def list_process_definitions(request: Request): """List all deployed process definitions""" return await proxy_request(request, "/process-definition") @router.get("/process-definition/{id}") async def get_process_definition(id: str, request: Request): """Get a specific process definition by ID""" return await proxy_request(request, f"/process-definition/{id}") @router.get("/process-definition/key/{key}") async def get_process_definition_by_key(key: str, request: Request): """Get the latest version of a process definition by key""" return await proxy_request(request, f"/process-definition/key/{key}") @router.get("/process-definition/{id}/xml") async def get_process_definition_xml(id: str, request: Request): """Get the BPMN 2.0 XML of a process definition""" return await proxy_request(request, f"/process-definition/{id}/xml") @router.get("/process-definition/key/{key}/xml") async def get_process_definition_xml_by_key(key: str, request: Request): """Get the BPMN 2.0 XML of a process definition by key""" return await proxy_request(request, f"/process-definition/key/{key}/xml") @router.get("/process-definition/{id}/diagram") async def get_process_definition_diagram(id: str, request: Request): """Get the diagram of a process definition""" return await proxy_request(request, f"/process-definition/{id}/diagram") # ============================================ # Process Instances # ============================================ @router.post("/process-definition/{id}/start") async def start_process_by_id(id: str, request: Request): """Start a new process instance by definition ID""" return await proxy_request(request, f"/process-definition/{id}/start") @router.post("/process-definition/key/{key}/start") async def start_process_by_key(key: str, request: Request): """Start a new process instance by definition key""" return await proxy_request(request, f"/process-definition/key/{key}/start") @router.get("/process-instance") async def list_process_instances(request: Request): """List all running process instances""" return await proxy_request(request, "/process-instance") @router.get("/process-instance/{id}") async def get_process_instance(id: str, request: Request): """Get a specific process instance""" return await proxy_request(request, f"/process-instance/{id}") @router.delete("/process-instance/{id}") async def delete_process_instance(id: str, request: Request): """Delete (cancel) a process instance""" return await proxy_request(request, f"/process-instance/{id}") @router.get("/process-instance/{id}/activity-instances") async def get_activity_instances(id: str, request: Request): """Get activity instances for a process instance""" return await proxy_request(request, f"/process-instance/{id}/activity-instances") # ============================================ # Tasks (User Tasks) # ============================================ @router.get("/task") async def list_tasks(request: Request): """List all pending tasks""" return await proxy_request(request, "/task") @router.get("/task/{id}") async def get_task(id: str, request: Request): """Get a specific task""" return await proxy_request(request, f"/task/{id}") @router.post("/task/{id}/claim") async def claim_task(id: str, request: Request): """Claim a task for a user""" return await proxy_request(request, f"/task/{id}/claim") @router.post("/task/{id}/unclaim") async def unclaim_task(id: str, request: Request): """Unclaim a task""" return await proxy_request(request, f"/task/{id}/unclaim") @router.post("/task/{id}/complete") async def complete_task(id: str, request: Request): """Complete a task with variables""" return await proxy_request(request, f"/task/{id}/complete") @router.post("/task/{id}/delegate") async def delegate_task(id: str, request: Request): """Delegate a task to another user""" return await proxy_request(request, f"/task/{id}/delegate") @router.get("/task/{id}/form-variables") async def get_task_form_variables(id: str, request: Request): """Get form variables for a task""" return await proxy_request(request, f"/task/{id}/form-variables") # ============================================ # Variables # ============================================ @router.get("/process-instance/{id}/variables") async def get_process_variables(id: str, request: Request): """Get all variables of a process instance""" return await proxy_request(request, f"/process-instance/{id}/variables") @router.put("/process-instance/{id}/variables") async def update_process_variables(id: str, request: Request): """Update variables of a process instance""" return await proxy_request(request, f"/process-instance/{id}/variables") @router.get("/task/{id}/variables") async def get_task_variables(id: str, request: Request): """Get all variables of a task""" return await proxy_request(request, f"/task/{id}/variables") # ============================================ # Deployment # ============================================ @router.get("/deployment") async def list_deployments(request: Request): """List all deployments""" return await proxy_request(request, "/deployment") @router.get("/deployment/{id}") async def get_deployment(id: str, request: Request): """Get a specific deployment""" return await proxy_request(request, f"/deployment/{id}") @router.delete("/deployment/{id}") async def delete_deployment(id: str, request: Request): """Delete a deployment""" return await proxy_request(request, f"/deployment/{id}") @router.post("/deployment/create") async def create_deployment( request: Request, deployment_name: str = Form(..., alias="deployment-name"), enable_duplicate_filtering: bool = Form(False, alias="enable-duplicate-filtering"), deploy_changed_only: bool = Form(False, alias="deploy-changed-only"), data: UploadFile = File(...) ): """ Deploy a BPMN process definition. Accepts multipart/form-data with: - deployment-name: Name for the deployment - data: The BPMN XML file """ url = f"{CAMUNDA_URL}/deployment/create" # Read the uploaded file file_content = await data.read() filename = data.filename or "process.bpmn" # Prepare multipart form data files = { "data": (filename, file_content, "application/octet-stream") } form_data = { "deployment-name": deployment_name, "enable-duplicate-filtering": str(enable_duplicate_filtering).lower(), "deploy-changed-only": str(deploy_changed_only).lower() } async with httpx.AsyncClient(timeout=60.0) as client: try: response = await client.post( url, files=files, data=form_data ) return Response( content=response.content, status_code=response.status_code, media_type="application/json" ) except httpx.ConnectError: raise HTTPException( status_code=503, detail="Camunda BPMN engine unavailable" ) except httpx.TimeoutException: raise HTTPException( status_code=504, detail="Deployment timeout" ) # ============================================ # History # ============================================ @router.get("/history/process-instance") async def list_historic_process_instances(request: Request): """List historic process instances""" return await proxy_request(request, "/history/process-instance") @router.get("/history/process-instance/{id}") async def get_historic_process_instance(id: str, request: Request): """Get a historic process instance""" return await proxy_request(request, f"/history/process-instance/{id}") @router.get("/history/activity-instance") async def list_historic_activity_instances(request: Request): """List historic activity instances""" return await proxy_request(request, "/history/activity-instance") @router.get("/history/task") async def list_historic_tasks(request: Request): """List historic tasks""" return await proxy_request(request, "/history/task") # ============================================ # Message Events # ============================================ @router.post("/message") async def correlate_message(request: Request): """Correlate a message to trigger a message event""" return await proxy_request(request, "/message") # ============================================ # Signal Events # ============================================ @router.post("/signal") async def throw_signal(request: Request): """Throw a signal to trigger signal events""" return await proxy_request(request, "/signal") # ============================================ # Convenience Endpoints (BreakPilot-specific) # ============================================ @router.get("/processes/active") async def get_active_processes(): """Get all active process instances with their current state""" async with httpx.AsyncClient(timeout=30.0) as client: try: # Get all running instances instances_response = await client.get(f"{CAMUNDA_URL}/process-instance") if instances_response.status_code != 200: return JSONResponse( content={"error": "Failed to fetch process instances"}, status_code=instances_response.status_code ) instances = instances_response.json() # Enrich with definition names result = [] for instance in instances: definition_id = instance.get("definitionId", "") try: def_response = await client.get( f"{CAMUNDA_URL}/process-definition/{definition_id}" ) if def_response.status_code == 200: definition = def_response.json() instance["processName"] = definition.get("name", definition.get("key", "Unknown")) except Exception: instance["processName"] = "Unknown" result.append(instance) return JSONResponse(content=result) except httpx.ConnectError: raise HTTPException(status_code=503, detail="Camunda unavailable") @router.get("/tasks/pending") async def get_pending_tasks(assignee: Optional[str] = None, candidateGroup: Optional[str] = None): """Get pending tasks, optionally filtered by assignee or candidate group""" params = {} if assignee: params["assignee"] = assignee if candidateGroup: params["candidateGroup"] = candidateGroup async with httpx.AsyncClient(timeout=30.0) as client: try: response = await client.get(f"{CAMUNDA_URL}/task", params=params) if response.status_code != 200: return JSONResponse( content={"error": "Failed to fetch tasks"}, status_code=response.status_code ) tasks = response.json() # Enrich tasks with process information for task in tasks: process_instance_id = task.get("processInstanceId") if process_instance_id: try: pi_response = await client.get( f"{CAMUNDA_URL}/process-instance/{process_instance_id}" ) if pi_response.status_code == 200: pi = pi_response.json() task["businessKey"] = pi.get("businessKey") except Exception: pass return JSONResponse(content=tasks) except httpx.ConnectError: raise HTTPException(status_code=503, detail="Camunda unavailable")