Repository: Significant-Gravitas/Auto-GPT
File: autogpt_platform/backend/backend/executor/utils.py (Python)

Code before:

class CancelExecutionEvent(BaseModel):
    graph_exec_id: str


async def _get_child_executions(
    db: "DatabaseManagerAsyncClient", parent_exec_id: str
) -> list["GraphExecution"]:
    """
    Get all child executions of a parent execution.

    Args:
        db: Database client
        parent_exec_id: Parent graph execution ID

    Returns:
        List of child graph executions
    """
    from backend.data.db import prisma

    if not prisma.is_connected():
        return []

    from backend.data.db import AgentGraphExecution

    children = await AgentGraphExecution.prisma().find_many(
        where={"parentGraphExecutionId": parent_exec_id}
    )

    from backend.data.execution import GraphExecution

    return [GraphExecution.from_db(child) for child in children]


async def stop_graph_execution(
    user_id: str,
    graph_exec_id: str,
    wait_timeout: float = 15.0,
    cascade: bool = True,
):
    """
    Stop a graph execution and optionally all its child executions.

    Mechanism:
    1. Set the cancel event for this execution
    2. If cascade=True, recursively stop all child executions
    3. Graph executor's cancel handler thread detects the event, terminates workers,
       reinitializes worker pool, and returns.
    4. Update execution statuses in DB and set `error` outputs to `"TERMINATED"`.

    Args:

Review comment (suggestion):
see  
`db = execution_db if prisma.is_connected() else get_database_manager_async_client()`

The same way we should do this, instead of doing direct query we should add the method on execution_db and access it using the model method or dbmanagerclient depending on the existence of prisma conneciton


Code after:
        routing_key="",  # not used for FANOUT
        durable=True,
        auto_delete=False,
    )
    return RabbitMQConfig(
        vhost="/",
        exchanges=[GRAPH_EXECUTION_EXCHANGE, GRAPH_EXECUTION_CANCEL_EXCHANGE],
        queues=[run_queue, cancel_queue],
    )


class CancelExecutionEvent(BaseModel):
    graph_exec_id: str


async def _get_child_executions(parent_exec_id: str) -> list["GraphExecutionMeta"]:
    """
    Get all child executions of a parent execution using the execution_db pattern.

    Args:
        parent_exec_id: Parent graph execution ID

    Returns:
        List of child graph executions
    """
    from backend.data.db import prisma

    if prisma.is_connected():
        edb = execution_db
    else:
        edb = get_database_manager_async_client()

    return await edb.get_child_graph_executions(parent_exec_id)


async def stop_graph_execution(
    user_id: str,
    graph_exec_id: str,
    wait_timeout: float = 15.0,
    cascade: bool = True,
):
    """
    Stop a graph execution and optionally all its child executions.

    Mechanism:
    1. Set the cancel event for this execution
    2. If cascade=True, recursively stop all child executions
    3. Graph executor's cancel handler thread detects the event, terminates workers,
       reinitializes worker pool, and returns.
    4. Update execution statuses in DB and set `error` outputs to `"TERMINATED"`.