mirror of
https://github.com/dograh-hq/dograh.git
synced 2026-06-25 08:48:13 +02:00
feat: add google stt and tts. add folders to organize agents
This commit is contained in:
Abhishek Kumar
parent
21951eca18
commit
ad2fa07058
52 changed files with 3412 additions and 621 deletions
99
api/routes/folder.py
Normal file
99
api/routes/folder.py
Normal file
|
|
@ -0,0 +1,99 @@
|
|||
from datetime import datetime
|
||||
|
||||
from fastapi import APIRouter, Depends, HTTPException
|
||||
from pydantic import BaseModel, Field, field_validator
|
||||
|
||||
from api.db import db_client
|
||||
from api.db.folder_client import FolderNameConflictError
|
||||
from api.db.models import UserModel
|
||||
from api.services.auth.depends import get_user
|
||||
|
||||
router = APIRouter(prefix="/folder")
|
||||
|
||||
|
||||
class FolderResponse(BaseModel):
|
||||
id: int
|
||||
name: str
|
||||
created_at: datetime
|
||||
|
||||
|
||||
class CreateFolderRequest(BaseModel):
|
||||
name: str = Field(..., min_length=1, max_length=100)
|
||||
|
||||
@field_validator("name")
|
||||
@classmethod
|
||||
def strip_name(cls, v: str) -> str:
|
||||
v = v.strip()
|
||||
if not v:
|
||||
raise ValueError("Folder name cannot be empty")
|
||||
return v
|
||||
|
||||
|
||||
class UpdateFolderRequest(CreateFolderRequest):
|
||||
pass
|
||||
|
||||
|
||||
@router.get("/")
|
||||
async def list_folders(
|
||||
user: UserModel = Depends(get_user),
|
||||
) -> list[FolderResponse]:
|
||||
"""List all folders in the authenticated user's organization."""
|
||||
folders = await db_client.list_folders(
|
||||
organization_id=user.selected_organization_id
|
||||
)
|
||||
return [
|
||||
FolderResponse(id=f.id, name=f.name, created_at=f.created_at) for f in folders
|
||||
]
|
||||
|
||||
|
||||
@router.post("/")
|
||||
async def create_folder(
|
||||
request: CreateFolderRequest,
|
||||
user: UserModel = Depends(get_user),
|
||||
) -> FolderResponse:
|
||||
"""Create a new folder in the authenticated user's organization."""
|
||||
try:
|
||||
folder = await db_client.create_folder(
|
||||
name=request.name,
|
||||
organization_id=user.selected_organization_id,
|
||||
)
|
||||
except FolderNameConflictError as e:
|
||||
raise HTTPException(status_code=409, detail=str(e))
|
||||
return FolderResponse(id=folder.id, name=folder.name, created_at=folder.created_at)
|
||||
|
||||
|
||||
@router.put("/{folder_id}")
|
||||
async def rename_folder(
|
||||
folder_id: int,
|
||||
request: UpdateFolderRequest,
|
||||
user: UserModel = Depends(get_user),
|
||||
) -> FolderResponse:
|
||||
"""Rename a folder owned by the authenticated user's organization."""
|
||||
try:
|
||||
folder = await db_client.rename_folder(
|
||||
folder_id=folder_id,
|
||||
name=request.name,
|
||||
organization_id=user.selected_organization_id,
|
||||
)
|
||||
except ValueError as e:
|
||||
raise HTTPException(status_code=404, detail=str(e))
|
||||
except FolderNameConflictError as e:
|
||||
raise HTTPException(status_code=409, detail=str(e))
|
||||
return FolderResponse(id=folder.id, name=folder.name, created_at=folder.created_at)
|
||||
|
||||
|
||||
@router.delete("/{folder_id}")
|
||||
async def delete_folder(
|
||||
folder_id: int,
|
||||
user: UserModel = Depends(get_user),
|
||||
) -> dict[str, bool]:
|
||||
"""Delete a folder. Member agents are moved to "Uncategorized", not deleted."""
|
||||
deleted = await db_client.delete_folder(
|
||||
folder_id=folder_id,
|
||||
organization_id=user.selected_organization_id,
|
||||
)
|
||||
if not deleted:
|
||||
raise HTTPException(
|
||||
status_code=404, detail=f"Folder with id {folder_id} not found"
|
||||
)
|
||||
return {"success": True}
|
||||
|
|
@ -6,6 +6,7 @@ from api.routes.agent_stream import router as agent_stream_router
|
|||
from api.routes.auth import router as auth_router
|
||||
from api.routes.campaign import router as campaign_router
|
||||
from api.routes.credentials import router as credentials_router
|
||||
from api.routes.folder import router as folder_router
|
||||
from api.routes.knowledge_base import router as knowledge_base_router
|
||||
from api.routes.node_types import router as node_types_router
|
||||
from api.routes.organization import router as organization_router
|
||||
|
|
@ -54,6 +55,7 @@ router.include_router(public_download_router)
|
|||
router.include_router(workflow_embed_router)
|
||||
router.include_router(knowledge_base_router)
|
||||
router.include_router(workflow_recording_router)
|
||||
router.include_router(folder_router)
|
||||
router.include_router(auth_router)
|
||||
router.include_router(node_types_router)
|
||||
router.include_router(agent_stream_router)
|
||||
|
|
|
|||
|
|
@ -1,18 +1,19 @@
|
|||
"""Public API endpoints for agent triggers.
|
||||
"""Public API endpoints for public agent execution.
|
||||
|
||||
These endpoints are accessible with API key authentication and allow
|
||||
external systems to programmatically trigger phone calls.
|
||||
"""
|
||||
|
||||
import random
|
||||
from typing import Optional
|
||||
from dataclasses import dataclass
|
||||
from typing import Awaitable, Callable, Optional
|
||||
|
||||
from fastapi import APIRouter, Header, HTTPException
|
||||
from loguru import logger
|
||||
from pydantic import BaseModel
|
||||
|
||||
from api.db import db_client
|
||||
from api.enums import TriggerState
|
||||
from api.enums import TriggerState, WorkflowStatus
|
||||
from api.services.quota_service import check_dograh_quota_by_user_id
|
||||
from api.services.telephony.factory import (
|
||||
get_default_telephony_provider,
|
||||
|
|
@ -39,6 +40,14 @@ class TriggerCallResponse(BaseModel):
|
|||
workflow_run_name: str
|
||||
|
||||
|
||||
@dataclass
|
||||
class ResolvedAgentTarget:
|
||||
workflow: object
|
||||
organization_id: int
|
||||
identifier_type: str
|
||||
identifier_value: str
|
||||
|
||||
|
||||
def trigger_exists_in_workflow(workflow_definition: dict, trigger_path: str) -> bool:
|
||||
"""Check if trigger node exists in workflow definition.
|
||||
|
||||
|
|
@ -57,72 +66,133 @@ def trigger_exists_in_workflow(workflow_definition: dict, trigger_path: str) ->
|
|||
return False
|
||||
|
||||
|
||||
async def _initiate_call(
|
||||
uuid: str,
|
||||
request: TriggerCallRequest,
|
||||
x_api_key: str,
|
||||
*,
|
||||
use_draft: bool,
|
||||
) -> TriggerCallResponse:
|
||||
"""Shared core for production and test trigger endpoints.
|
||||
|
||||
When ``use_draft`` is True the latest draft definition is executed;
|
||||
otherwise the published (released) definition is used.
|
||||
"""
|
||||
# 1. Validate API key
|
||||
async def _validate_api_key(x_api_key: str):
|
||||
"""Validate the org API key used to invoke a public agent endpoint."""
|
||||
api_key = await db_client.validate_api_key(x_api_key)
|
||||
if not api_key:
|
||||
raise HTTPException(status_code=401, detail="Invalid API key")
|
||||
return api_key
|
||||
|
||||
# 2. Lookup agent trigger by UUID
|
||||
trigger = await db_client.get_agent_trigger_by_path(uuid)
|
||||
|
||||
def _ensure_workflow_is_active(workflow) -> None:
|
||||
if workflow.status != WorkflowStatus.ACTIVE.value:
|
||||
raise HTTPException(status_code=404, detail="Workflow is not active")
|
||||
|
||||
|
||||
def _get_execution_user_id(workflow) -> int:
|
||||
if workflow.user_id is None:
|
||||
raise HTTPException(
|
||||
status_code=409,
|
||||
detail="Workflow has no execution owner",
|
||||
)
|
||||
return workflow.user_id
|
||||
|
||||
|
||||
async def _get_workflow_definition_for_execution(workflow, *, use_draft: bool) -> dict:
|
||||
"""Return the definition that would execute for this public agent request."""
|
||||
if use_draft:
|
||||
draft = await db_client.get_draft_version(workflow.id)
|
||||
if draft:
|
||||
return draft.workflow_json
|
||||
|
||||
if workflow.released_definition is None:
|
||||
raise HTTPException(
|
||||
status_code=404, detail="Workflow has no published definition"
|
||||
)
|
||||
|
||||
return workflow.released_definition.workflow_json
|
||||
|
||||
|
||||
async def _resolve_trigger_target(
|
||||
trigger_path: str,
|
||||
organization_id: int,
|
||||
*,
|
||||
use_draft: bool,
|
||||
) -> ResolvedAgentTarget:
|
||||
"""Resolve a trigger UUID to a workflow, scoped to the API key's org."""
|
||||
trigger = await db_client.get_agent_trigger_by_path(trigger_path)
|
||||
if not trigger:
|
||||
raise HTTPException(status_code=404, detail="Agent trigger not found")
|
||||
|
||||
# 3. Validate organization match (API key org must match trigger org)
|
||||
if api_key.organization_id != trigger.organization_id:
|
||||
if organization_id != trigger.organization_id:
|
||||
raise HTTPException(status_code=403, detail="Access denied")
|
||||
|
||||
# 4. Validate trigger is active
|
||||
if trigger.state != TriggerState.ACTIVE.value:
|
||||
raise HTTPException(status_code=404, detail="Agent trigger is not active")
|
||||
|
||||
# 4.5 Check Dograh quota before initiating the call (apply the trigger's
|
||||
# workflow's model_overrides so we evaluate the keys this run will use).
|
||||
workflow = await db_client.get_workflow(
|
||||
trigger.workflow_id,
|
||||
organization_id=organization_id,
|
||||
)
|
||||
if not workflow:
|
||||
raise HTTPException(status_code=404, detail="Workflow not found")
|
||||
|
||||
_ensure_workflow_is_active(workflow)
|
||||
workflow_definition = await _get_workflow_definition_for_execution(
|
||||
workflow,
|
||||
use_draft=use_draft,
|
||||
)
|
||||
if not trigger_exists_in_workflow(workflow_definition, trigger_path):
|
||||
raise HTTPException(
|
||||
status_code=404,
|
||||
detail="Trigger not found in the selected Agent",
|
||||
)
|
||||
|
||||
return ResolvedAgentTarget(
|
||||
workflow=workflow,
|
||||
organization_id=organization_id,
|
||||
identifier_type="trigger_path",
|
||||
identifier_value=trigger_path,
|
||||
)
|
||||
|
||||
|
||||
async def _resolve_workflow_uuid_target(
|
||||
workflow_uuid: str,
|
||||
organization_id: int,
|
||||
*,
|
||||
use_draft: bool,
|
||||
) -> ResolvedAgentTarget:
|
||||
"""Resolve a workflow UUID directly, scoped to the API key's org."""
|
||||
workflow = await db_client.get_workflow_by_uuid(workflow_uuid, organization_id)
|
||||
if not workflow:
|
||||
raise HTTPException(status_code=404, detail="Workflow not found")
|
||||
|
||||
_ensure_workflow_is_active(workflow)
|
||||
await _get_workflow_definition_for_execution(workflow, use_draft=use_draft)
|
||||
|
||||
return ResolvedAgentTarget(
|
||||
workflow=workflow,
|
||||
organization_id=organization_id,
|
||||
identifier_type="workflow_uuid",
|
||||
identifier_value=workflow_uuid,
|
||||
)
|
||||
|
||||
|
||||
async def _execute_resolved_target(
|
||||
target: ResolvedAgentTarget,
|
||||
request: TriggerCallRequest,
|
||||
*,
|
||||
use_draft: bool,
|
||||
api_key_id: int | None,
|
||||
api_key_created_by: int | None,
|
||||
) -> TriggerCallResponse:
|
||||
"""Shared execution path once the target workflow has been resolved."""
|
||||
execution_user_id = _get_execution_user_id(target.workflow)
|
||||
|
||||
# Check Dograh quota using the workflow owner's config and model overrides.
|
||||
quota_result = await check_dograh_quota_by_user_id(
|
||||
api_key.created_by, workflow_id=trigger.workflow_id
|
||||
execution_user_id,
|
||||
workflow_id=target.workflow.id,
|
||||
)
|
||||
if not quota_result.has_quota:
|
||||
raise HTTPException(status_code=402, detail=quota_result.error_message)
|
||||
|
||||
# 5. Get workflow and resolve the definition (published vs draft)
|
||||
workflow = await db_client.get_workflow_by_id(trigger.workflow_id)
|
||||
if not workflow:
|
||||
raise HTTPException(status_code=404, detail="Workflow not found")
|
||||
|
||||
if use_draft:
|
||||
draft = await db_client.get_draft_version(trigger.workflow_id)
|
||||
# Fall back to the published definition when no draft exists, so the
|
||||
# test URL always runs *something* — typically the same agent the
|
||||
# production URL would run.
|
||||
workflow_definition = (
|
||||
draft.workflow_json if draft else workflow.released_definition.workflow_json
|
||||
)
|
||||
else:
|
||||
workflow_definition = workflow.released_definition.workflow_json
|
||||
|
||||
# Validate trigger node still exists in the resolved definition
|
||||
if not trigger_exists_in_workflow(workflow_definition, uuid):
|
||||
raise HTTPException(
|
||||
status_code=404,
|
||||
detail="Trigger not found in the published Agent",
|
||||
)
|
||||
|
||||
# 6. Get telephony provider — either the caller-specified config (validated
|
||||
# against the trigger's org) or the org's default config.
|
||||
# Get telephony provider — either the caller-specified config (validated
|
||||
# against the workflow's org) or the org's default config.
|
||||
if request.telephony_configuration_id is not None:
|
||||
cfg = await db_client.get_telephony_configuration_for_org(
|
||||
request.telephony_configuration_id, trigger.organization_id
|
||||
request.telephony_configuration_id,
|
||||
target.organization_id,
|
||||
)
|
||||
if not cfg:
|
||||
raise HTTPException(
|
||||
|
|
@ -130,7 +200,7 @@ async def _initiate_call(
|
|||
)
|
||||
try:
|
||||
provider = await get_telephony_provider_by_id(
|
||||
cfg.id, trigger.organization_id
|
||||
cfg.id, target.organization_id
|
||||
)
|
||||
except ValueError:
|
||||
raise HTTPException(
|
||||
|
|
@ -140,14 +210,14 @@ async def _initiate_call(
|
|||
resolved_cfg_id = cfg.id
|
||||
else:
|
||||
try:
|
||||
provider = await get_default_telephony_provider(trigger.organization_id)
|
||||
provider = await get_default_telephony_provider(target.organization_id)
|
||||
except ValueError:
|
||||
raise HTTPException(
|
||||
status_code=400,
|
||||
detail="Telephony provider not configured for this organization",
|
||||
)
|
||||
default_cfg = await db_client.get_default_telephony_configuration(
|
||||
trigger.organization_id
|
||||
target.organization_id
|
||||
)
|
||||
resolved_cfg_id = default_cfg.id if default_cfg else None
|
||||
|
||||
|
|
@ -164,24 +234,36 @@ async def _initiate_call(
|
|||
# 8. Create workflow run
|
||||
mode_label = "TEST" if use_draft else "API"
|
||||
workflow_run_name = f"WR-{mode_label}-{random.randint(1000, 9999)}"
|
||||
initial_context = {
|
||||
"provider": provider.PROVIDER_NAME,
|
||||
"phone_number": request.phone_number,
|
||||
"trigger_mode": "test" if use_draft else "production",
|
||||
"telephony_configuration_id": resolved_cfg_id,
|
||||
"agent_identifier": target.identifier_value,
|
||||
"agent_identifier_type": target.identifier_type,
|
||||
"workflow_uuid": target.workflow.workflow_uuid,
|
||||
}
|
||||
if target.identifier_type == "trigger_path":
|
||||
initial_context["agent_uuid"] = target.identifier_value
|
||||
if api_key_id is not None:
|
||||
initial_context["api_key_id"] = api_key_id
|
||||
if api_key_created_by is not None:
|
||||
initial_context["api_key_created_by"] = api_key_created_by
|
||||
initial_context.update(request.initial_context or {})
|
||||
|
||||
workflow_run = await db_client.create_workflow_run(
|
||||
name=workflow_run_name,
|
||||
workflow_id=trigger.workflow_id,
|
||||
workflow_id=target.workflow.id,
|
||||
mode=workflow_run_mode,
|
||||
initial_context={
|
||||
"provider": provider.PROVIDER_NAME,
|
||||
"phone_number": request.phone_number,
|
||||
"agent_uuid": uuid,
|
||||
"trigger_mode": "test" if use_draft else "production",
|
||||
"telephony_configuration_id": resolved_cfg_id,
|
||||
**(request.initial_context or {}),
|
||||
},
|
||||
user_id=api_key.created_by,
|
||||
initial_context=initial_context,
|
||||
user_id=execution_user_id,
|
||||
use_draft=use_draft,
|
||||
organization_id=target.organization_id,
|
||||
)
|
||||
|
||||
logger.info(
|
||||
f"Created workflow run {workflow_run.id} for API trigger {uuid} "
|
||||
f"Created workflow run {workflow_run.id} for public agent "
|
||||
f"{target.identifier_type}={target.identifier_value} "
|
||||
f"(mode={'test' if use_draft else 'production'}) "
|
||||
f"to phone number {request.phone_number}"
|
||||
)
|
||||
|
|
@ -192,10 +274,10 @@ async def _initiate_call(
|
|||
|
||||
webhook_url = (
|
||||
f"{backend_endpoint}/api/v1/telephony/{webhook_endpoint}"
|
||||
f"?workflow_id={trigger.workflow_id}"
|
||||
f"&user_id={api_key.created_by}"
|
||||
f"?workflow_id={target.workflow.id}"
|
||||
f"&user_id={execution_user_id}"
|
||||
f"&workflow_run_id={workflow_run.id}"
|
||||
f"&organization_id={trigger.organization_id}"
|
||||
f"&organization_id={target.organization_id}"
|
||||
)
|
||||
|
||||
# 10. Initiate call via telephony provider. workflow_id and user_id are
|
||||
|
|
@ -207,8 +289,8 @@ async def _initiate_call(
|
|||
to_number=request.phone_number,
|
||||
webhook_url=webhook_url,
|
||||
workflow_run_id=workflow_run.id,
|
||||
workflow_id=trigger.workflow_id,
|
||||
user_id=api_key.created_by,
|
||||
workflow_id=target.workflow.id,
|
||||
user_id=execution_user_id,
|
||||
)
|
||||
except Exception as e:
|
||||
logger.warning(
|
||||
|
|
@ -221,7 +303,7 @@ async def _initiate_call(
|
|||
|
||||
logger.info(
|
||||
f"Call initiated successfully for workflow run {workflow_run.id} "
|
||||
f"via trigger {uuid}"
|
||||
f"via {target.identifier_type}={target.identifier_value}"
|
||||
)
|
||||
|
||||
return TriggerCallResponse(
|
||||
|
|
@ -231,6 +313,30 @@ async def _initiate_call(
|
|||
)
|
||||
|
||||
|
||||
async def _initiate_call(
|
||||
identifier: str,
|
||||
request: TriggerCallRequest,
|
||||
x_api_key: str,
|
||||
*,
|
||||
use_draft: bool,
|
||||
target_resolver: Callable[..., Awaitable[ResolvedAgentTarget]],
|
||||
) -> TriggerCallResponse:
|
||||
"""Resolve the requested public target, then execute the common call flow."""
|
||||
api_key = await _validate_api_key(x_api_key)
|
||||
target = await target_resolver(
|
||||
identifier,
|
||||
api_key.organization_id,
|
||||
use_draft=use_draft,
|
||||
)
|
||||
return await _execute_resolved_target(
|
||||
target,
|
||||
request,
|
||||
use_draft=use_draft,
|
||||
api_key_id=api_key.id,
|
||||
api_key_created_by=api_key.created_by,
|
||||
)
|
||||
|
||||
|
||||
@router.post("/{uuid}", response_model=TriggerCallResponse)
|
||||
async def initiate_call(
|
||||
uuid: str,
|
||||
|
|
@ -241,7 +347,13 @@ async def initiate_call(
|
|||
|
||||
Executes the workflow's currently released definition.
|
||||
"""
|
||||
return await _initiate_call(uuid, request, x_api_key, use_draft=False)
|
||||
return await _initiate_call(
|
||||
uuid,
|
||||
request,
|
||||
x_api_key,
|
||||
use_draft=False,
|
||||
target_resolver=_resolve_trigger_target,
|
||||
)
|
||||
|
||||
|
||||
@router.post("/test/{uuid}", response_model=TriggerCallResponse)
|
||||
|
|
@ -255,4 +367,42 @@ async def initiate_call_test(
|
|||
Useful for verifying changes before publishing. Falls back to the
|
||||
published definition when no draft exists.
|
||||
"""
|
||||
return await _initiate_call(uuid, request, x_api_key, use_draft=True)
|
||||
return await _initiate_call(
|
||||
uuid,
|
||||
request,
|
||||
x_api_key,
|
||||
use_draft=True,
|
||||
target_resolver=_resolve_trigger_target,
|
||||
)
|
||||
|
||||
|
||||
@router.post("/workflow/{workflow_uuid}", response_model=TriggerCallResponse)
|
||||
async def initiate_call_by_workflow_uuid(
|
||||
workflow_uuid: str,
|
||||
request: TriggerCallRequest,
|
||||
x_api_key: str = Header(..., alias="X-API-Key"),
|
||||
):
|
||||
"""Initiate a phone call against the published workflow identified by UUID."""
|
||||
return await _initiate_call(
|
||||
workflow_uuid,
|
||||
request,
|
||||
x_api_key,
|
||||
use_draft=False,
|
||||
target_resolver=_resolve_workflow_uuid_target,
|
||||
)
|
||||
|
||||
|
||||
@router.post("/test/workflow/{workflow_uuid}", response_model=TriggerCallResponse)
|
||||
async def initiate_call_test_by_workflow_uuid(
|
||||
workflow_uuid: str,
|
||||
request: TriggerCallRequest,
|
||||
x_api_key: str = Header(..., alias="X-API-Key"),
|
||||
):
|
||||
"""Initiate a phone call against the latest draft of the workflow by UUID."""
|
||||
return await _initiate_call(
|
||||
workflow_uuid,
|
||||
request,
|
||||
x_api_key,
|
||||
use_draft=True,
|
||||
target_resolver=_resolve_workflow_uuid_target,
|
||||
)
|
||||
|
|
|
|||
|
|
@ -305,7 +305,9 @@ async def _validate_inbound_request(
|
|||
"""
|
||||
from api.services.telephony import registry as telephony_registry
|
||||
|
||||
workflow = await db_client.get_workflow(workflow_id)
|
||||
# System lookup: inbound routing only has the workflow_id and derives the
|
||||
# org/user from the workflow itself, so use the explicit unscoped variant.
|
||||
workflow = await db_client.get_workflow_by_id(workflow_id)
|
||||
if not workflow:
|
||||
return False, TelephonyError.WORKFLOW_NOT_FOUND, {}, None
|
||||
|
||||
|
|
@ -528,8 +530,9 @@ async def _handle_telephony_websocket(
|
|||
await websocket.close(code=4404, reason="Workflow run not found")
|
||||
return
|
||||
|
||||
# Get workflow for organization info
|
||||
workflow = await db_client.get_workflow(workflow_id)
|
||||
# Get workflow for organization info. System lookup keyed only on the
|
||||
# workflow_id (org is derived below) — use the explicit unscoped variant.
|
||||
workflow = await db_client.get_workflow_by_id(workflow_id)
|
||||
if not workflow:
|
||||
logger.error(f"Workflow {workflow_id} not found")
|
||||
await websocket.close(code=4404, reason="Workflow not found")
|
||||
|
|
|
|||
|
|
@ -32,99 +32,16 @@ from api.services.storage import storage_fs
|
|||
from api.services.workflow.dto import ReactFlowDTO, sanitize_workflow_definition
|
||||
from api.services.workflow.duplicate import duplicate_workflow
|
||||
from api.services.workflow.errors import ItemKind, WorkflowError
|
||||
from api.services.workflow.trigger_paths import (
|
||||
TriggerPathIssue,
|
||||
ensure_trigger_paths,
|
||||
extract_trigger_paths,
|
||||
regenerate_trigger_uuids,
|
||||
trigger_path_to_node_id,
|
||||
validate_trigger_paths,
|
||||
)
|
||||
from api.services.workflow.workflow_graph import WorkflowGraph
|
||||
|
||||
|
||||
def extract_trigger_paths(workflow_definition: dict) -> List[str]:
|
||||
"""Extract trigger UUIDs from workflow definition.
|
||||
|
||||
Args:
|
||||
workflow_definition: The workflow definition JSON
|
||||
|
||||
Returns:
|
||||
List of trigger UUIDs found in the workflow
|
||||
"""
|
||||
if not workflow_definition:
|
||||
return []
|
||||
|
||||
nodes = workflow_definition.get("nodes", [])
|
||||
trigger_paths = []
|
||||
|
||||
for node in nodes:
|
||||
if node.get("type") == "trigger":
|
||||
trigger_path = node.get("data", {}).get("trigger_path")
|
||||
if trigger_path:
|
||||
trigger_paths.append(trigger_path)
|
||||
|
||||
return trigger_paths
|
||||
|
||||
|
||||
def _trigger_path_to_node_id(workflow_definition: dict) -> dict[str, str]:
|
||||
"""Map each trigger node's trigger_path to its node id."""
|
||||
if not workflow_definition:
|
||||
return {}
|
||||
out: dict[str, str] = {}
|
||||
for node in workflow_definition.get("nodes", []):
|
||||
if node.get("type") == "trigger":
|
||||
tp = node.get("data", {}).get("trigger_path")
|
||||
if tp:
|
||||
out[tp] = node.get("id")
|
||||
return out
|
||||
|
||||
|
||||
def regenerate_trigger_uuids(workflow_definition: dict) -> dict:
|
||||
"""Regenerate UUIDs for all trigger nodes in a workflow definition.
|
||||
|
||||
This should be called when creating a new workflow from a template or
|
||||
duplicating a workflow to avoid trigger UUID conflicts.
|
||||
|
||||
Args:
|
||||
workflow_definition: The workflow definition JSON
|
||||
|
||||
Returns:
|
||||
Updated workflow definition with new trigger UUIDs
|
||||
"""
|
||||
if not workflow_definition:
|
||||
return workflow_definition
|
||||
|
||||
# Deep copy to avoid modifying the original
|
||||
import copy
|
||||
|
||||
updated_definition = copy.deepcopy(workflow_definition)
|
||||
|
||||
nodes = updated_definition.get("nodes", [])
|
||||
for node in nodes:
|
||||
if node.get("type") == "trigger":
|
||||
# Generate a new UUID for this trigger
|
||||
if "data" not in node:
|
||||
node["data"] = {}
|
||||
node["data"]["trigger_path"] = str(uuid.uuid4())
|
||||
|
||||
return updated_definition
|
||||
|
||||
|
||||
def ensure_trigger_paths(workflow_definition: Optional[dict]) -> Optional[dict]:
|
||||
"""Mint a UUID for any trigger node that's missing ``data.trigger_path``.
|
||||
|
||||
Trigger nodes that already carry a non-empty trigger_path are left
|
||||
untouched so stable IDs survive edits. The input is not mutated; the
|
||||
returned dict is what should be persisted and echoed in the response.
|
||||
"""
|
||||
if not workflow_definition:
|
||||
return workflow_definition
|
||||
|
||||
import copy
|
||||
|
||||
out = copy.deepcopy(workflow_definition)
|
||||
for node in out.get("nodes") or []:
|
||||
if node.get("type") != "trigger":
|
||||
continue
|
||||
data = node.setdefault("data", {})
|
||||
if not data.get("trigger_path"):
|
||||
data["trigger_path"] = str(uuid.uuid4())
|
||||
return out
|
||||
|
||||
|
||||
router = APIRouter(prefix="/workflow")
|
||||
|
||||
|
||||
|
|
@ -139,7 +56,7 @@ def _trigger_conflict_http_exception(
|
|||
"""Build a 409 with the same detail shape as validate's 422 so the editor
|
||||
can highlight the offending trigger node(s) using the same code path."""
|
||||
path_to_node = (
|
||||
_trigger_path_to_node_id(workflow_definition) if workflow_definition else {}
|
||||
trigger_path_to_node_id(workflow_definition) if workflow_definition else {}
|
||||
)
|
||||
errors: list[WorkflowError] = [
|
||||
WorkflowError(
|
||||
|
|
@ -159,6 +76,24 @@ def _trigger_conflict_http_exception(
|
|||
)
|
||||
|
||||
|
||||
def _trigger_path_validation_http_exception(
|
||||
issues: list[TriggerPathIssue],
|
||||
) -> HTTPException:
|
||||
errors = [
|
||||
WorkflowError(
|
||||
kind=ItemKind.node,
|
||||
id=issue.node_id,
|
||||
field="data.trigger_path",
|
||||
message=issue.message,
|
||||
)
|
||||
for issue in issues
|
||||
]
|
||||
return HTTPException(
|
||||
status_code=422,
|
||||
detail=ValidateWorkflowResponse(is_valid=False, errors=errors).model_dump(),
|
||||
)
|
||||
|
||||
|
||||
async def _validate_workflow_definition(
|
||||
workflow_definition: Optional[dict],
|
||||
exclude_workflow_id: Optional[int] = None,
|
||||
|
|
@ -187,6 +122,17 @@ async def _validate_workflow_definition(
|
|||
except ValueError as e:
|
||||
errors.extend(e.args[0])
|
||||
|
||||
# ----------- Trigger Path Format Check ------------
|
||||
for issue in validate_trigger_paths(workflow_definition):
|
||||
errors.append(
|
||||
WorkflowError(
|
||||
kind=ItemKind.node,
|
||||
id=issue.node_id,
|
||||
field="data.trigger_path",
|
||||
message=issue.message,
|
||||
)
|
||||
)
|
||||
|
||||
# ----------- Trigger Path Conflict Check ------------
|
||||
trigger_paths = extract_trigger_paths(workflow_definition)
|
||||
if trigger_paths:
|
||||
|
|
@ -195,7 +141,7 @@ async def _validate_workflow_definition(
|
|||
exclude_workflow_id=exclude_workflow_id,
|
||||
)
|
||||
if conflicts:
|
||||
path_to_node = _trigger_path_to_node_id(workflow_definition)
|
||||
path_to_node = trigger_path_to_node_id(workflow_definition)
|
||||
for conflicting_path in conflicts:
|
||||
errors.append(
|
||||
WorkflowError(
|
||||
|
|
@ -251,6 +197,14 @@ class WorkflowListResponse(BaseModel):
|
|||
status: str
|
||||
created_at: datetime
|
||||
total_runs: int
|
||||
folder_id: int | None = None
|
||||
workflow_uuid: str | None = None
|
||||
|
||||
|
||||
class MoveWorkflowToFolderRequest(BaseModel):
|
||||
"""Move a workflow into a folder, or to "Uncategorized" when null."""
|
||||
|
||||
folder_id: int | None = None
|
||||
|
||||
|
||||
class WorkflowCountResponse(BaseModel):
|
||||
|
|
@ -404,6 +358,9 @@ async def create_workflow(
|
|||
# Auto-mint trigger_path for any trigger node that didn't ship one so
|
||||
# clients don't need to generate UUIDs themselves.
|
||||
workflow_definition = ensure_trigger_paths(request.workflow_definition)
|
||||
trigger_path_issues = validate_trigger_paths(workflow_definition)
|
||||
if trigger_path_issues:
|
||||
raise _trigger_path_validation_http_exception(trigger_path_issues)
|
||||
|
||||
# Validate trigger path uniqueness BEFORE creating the workflow so we
|
||||
# don't leave an orphaned workflow record when the trigger conflicts.
|
||||
|
|
@ -641,6 +598,8 @@ async def get_workflows(
|
|||
status=workflow.status,
|
||||
created_at=workflow.created_at,
|
||||
total_runs=run_counts.get(workflow.id, 0),
|
||||
folder_id=workflow.folder_id,
|
||||
workflow_uuid=workflow.workflow_uuid,
|
||||
)
|
||||
for workflow in workflows
|
||||
]
|
||||
|
|
@ -883,6 +842,48 @@ async def update_workflow_status(
|
|||
raise HTTPException(status_code=500, detail=str(e))
|
||||
|
||||
|
||||
@router.put("/{workflow_id}/folder")
|
||||
async def move_workflow_to_folder(
|
||||
workflow_id: int,
|
||||
request: MoveWorkflowToFolderRequest,
|
||||
user: UserModel = Depends(get_user),
|
||||
) -> WorkflowListResponse:
|
||||
"""Move a workflow into a folder, or to "Uncategorized" (folder_id=null).
|
||||
|
||||
Validates that the target folder belongs to the caller's organization —
|
||||
the FK alone proves the folder exists, not that the caller may use it.
|
||||
"""
|
||||
# Validate target folder ownership (tenant isolation) unless un-filing.
|
||||
if request.folder_id is not None:
|
||||
folder = await db_client.get_folder(
|
||||
request.folder_id, organization_id=user.selected_organization_id
|
||||
)
|
||||
if folder is None:
|
||||
raise HTTPException(
|
||||
status_code=404,
|
||||
detail=f"Folder with id {request.folder_id} not found",
|
||||
)
|
||||
|
||||
try:
|
||||
workflow = await db_client.move_workflow_to_folder(
|
||||
workflow_id=workflow_id,
|
||||
folder_id=request.folder_id,
|
||||
organization_id=user.selected_organization_id,
|
||||
)
|
||||
except ValueError as e:
|
||||
raise HTTPException(status_code=404, detail=str(e))
|
||||
|
||||
run_count = await db_client.get_workflow_run_count(workflow.id)
|
||||
return WorkflowListResponse(
|
||||
id=workflow.id,
|
||||
name=workflow.name,
|
||||
status=workflow.status,
|
||||
created_at=workflow.created_at,
|
||||
total_runs=run_count,
|
||||
folder_id=workflow.folder_id,
|
||||
)
|
||||
|
||||
|
||||
@router.put(
|
||||
"/{workflow_id}",
|
||||
**sdk_expose(
|
||||
|
|
@ -917,6 +918,9 @@ async def update_workflow(
|
|||
# response echoes workflow_definition so the client picks up the new
|
||||
# UUID without a refetch.
|
||||
workflow_definition = ensure_trigger_paths(workflow_definition)
|
||||
trigger_path_issues = validate_trigger_paths(workflow_definition)
|
||||
if trigger_path_issues:
|
||||
raise _trigger_path_validation_http_exception(trigger_path_issues)
|
||||
if workflow_definition:
|
||||
existing_workflow = await db_client.get_workflow(
|
||||
workflow_id, organization_id=user.selected_organization_id
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue