update_workflow_stages

imerit_ango.sdk.SDK.

update_workflow_stages(project_id, stages)

Update the workflow of your project.

Parameters

• project_id: string

  • The unique identifier for the project. You can find the project ID in the user interface or retrieve it using the list_projects function.

• stages: List[dict]

  • The workflow of the project, in JSON format.

  • Stage types and their required fields:

Start Stage

• "id": "Start"

• "type": "Start"

• "name": "Start"

• "position": {"x": 10, "y": 20}

• "next": ["<NEXT STAGE ID>"]

• "autoForward": true or false

Complete Stage

• "id": "Complete"

• "type": "Complete"

• "name": "Complete"

• "position": {"x": 10, "y": 20}

• "preventRequeue": true or false

Label Stage

• "id": "<STAGE ID>"

• "type": "Label"

• "name": "Label"

• "position": {"x": 10, "y": 20}

• "next": ["<NEXT STAGE ID>"]

• "assignedTo": ["[e-mail]", "[e-mail]"]

Consensus Stage

• "id": "<CONSENSUS STAGE ID>"

• "type": "Consensus"

• "name": "Consensus"

• "position": {"x": 10, "y": 20}

• "next": ["<NEXT STAGE ID AGREEMENT>", "<NEXT STAGE ID DISAGREEMENT>"]

• "consensusId": null

• "consensusConfig": {"threshold": {}, "adjudicationMethod": "best"}

Two types of stages, “Label” and “Plugin”, can be added to a consensus stage. They should adhere to the following format:

Label Stage:

• "id": "<STAGE ID>"

• "type": "Label"

• "name": "Consensus_1"

• "position": {"x": 0, "y": 0}

• "next": ["<CONSENSUS STAGE ID>"]

• "assignedTo": ["[e-mail]", "[e-mail]"]

• "consensusId": "<CONSENSUS STAGE ID>"

• "consensusConfig": {}

Plugin Stage:

• "id": "<STAGE ID>"

• "type": "Plugin"

• "name": "Consensus_2"

• "position": {"x": 0, "y": 0}

• "next": ["<CONSENSUS STAGE ID>"]

• "pluginId": "<PLUGIN ID>"

• "pluginConfig": {"categorySchema": [], "configJSON": "", "overwrite": true}

• "consensusId": "<CONSENSUS STAGE ID>"

Review Stage

• "id": "<STAGE ID>"

• "type": "Review"

• "name": "Review"

• "position": {"x": 10, "y": 20}

• "next": ["<NEXT STAGE ID ACCEPTED>", "<NEXT STAGE ID REJECTED>"]

• "assignedTo": ["[e-mail]", "[e-mail]"]

• "readOnly": true or false

Hold Stage

• "id": "<STAGE ID>"

• "type": "Hold"

• "name": "Hold"

• "position": {"x": 10, "y": 20}

• "next": ["<NEXT STAGE ID>"]

Logic Stage

• "id": "<STAGE ID>"

• "type": "Logic"

• "name": "Logic"

• "position": {"x": 10, "y": 20}

• "next": ["<NEXT STAGE ID IF CONDITION>", ..., "<NEXT STAGE ID ELSE CONDITION>"]

• "logic": {"conditions": [...]}

• A single condition should follow:

Logic Type: Annotation Type

• {"type": "AnnotationType", "value": {"tools": [{"schemaId": "<Schema ID>", "filter": ["exists"]}]}, "index": 0}

Logic Type: Assignee

• {"type": "Annotator", "value": ["[e-mail]"], "index": 0}

Logic Type: Random Sample

• {"type": "RandomSample", "value": 0.100", "index": 0}

Logic Type: Task Duration

• {"type": "Duration", "value": [Start Time, End Time], "index": 0}

Logic Type: Batch

• {"type": "Batch", "value": ["<SELECTED BATCH ID>"], "index": 0}

Plugin Stage

• "id": "<STAGE ID>"

• "type": "Plugin"

• "name": "Plugin"

• "position": {"x": 10, "y": 20}

• "next": ["<NEXT STAGE ID>"]

• "pluginId": "<PLUGIN ID>"

• "pluginConfig": {"overwrite": true, "categorySchema": [], "configJSON": ""}

Webhook Stage

• "id": "<STAGE ID>"

• "type": "Webhook"

• "name": "Webhook"

• "position": {"x": 10, "y": 20}

• "next": ["<NEXT STAGE ID>"]

• "webhook": {"authType": "<Auth Type>", "secret": "<SECRET>", "url": "<URL>"}

Note: To get an idea of what you can pass as input in Stages, you may create a workflow in a project, then run:

ango_sdk.get_project(<PROJECT ID>).get("data").get("project").get("stages")

As the returned JSON, you'll get the JSON representation of the project's workflow, which you can use as a skeleton to create your own workflow JSONs.

Returns:

• output: dict

  • A dictionary containing the result of the operation.

  • Including a status field indicating whether the request was successful and a data field containing the response payload with updated resources produced by the operation.

How to verify in Ango Hub?

After successfully executing the update_workflow_stages function, you can validate the changes directly in Ango Hub.

Navigate to: Projects → [Your Project] → Workflow

• Confirm that the workflow stages have been updated as expected.

Changes made via the SDK are reflected in Ango Hub in near real-time. If updates are not immediately visible, please refresh the page.

Example

Create a workflow from scratch

python

import os
from dotenv import load_dotenv
from imerit_ango.sdk import SDK

load_dotenv('variables.env')
api_key = os.getenv('API_KEY')
project_id = os.getenv('PROJECT_ID')

ango_sdk = SDK(api_key)

start_stage = {"id": "Start",
               "type": "Start",
               "name": "Start",
               "next": ["Label"],
               "autoForward": False,
               "position": {"x": 0, "y": 0}}

label_stage = {"id": "Label",
               "type": "Label",
               "name": "Label",
               "next": ["Review"],
               "assignedTo": [],
               "position": {"x": 400, "y": 0}}

review_stage = {"id": "Review",
                "type": "Review",
                "name": "Review",
                "next": ["Complete", "Label"],
                "assignedTo": [],
                "position": {"x": 800, "y": 0}}

complete_stage = {"id": "Complete",
                  "type": "Complete",
                  "name": "Complete",
                  "next": [],
                  "position": {"x": 1200, "y": 0},
                  "preventRequeue": False}

stages = [start_stage, label_stage, review_stage, complete_stage]

sdk_response = ango_sdk.update_workflow_stages(project_id=project_id, stages=stages)

curl

Disable the "Auto Forward on Upload" option on the Start stage

python

import os
from dotenv import load_dotenv
from imerit_ango.sdk import SDK

load_dotenv('variables.env')
api_key = os.getenv('API_KEY')
project_id = os.getenv('PROJECT_ID')

ango_sdk = SDK(api_key)

stages = ango_sdk.get_project(project_id=project_id).get("data").get("project").get("stages")

new_stages = []
for stage in stages:
    new_stage = stage.copy()
    if stage["id"] == "Start":
        new_stage["autoForward"] = False
    new_stages.append(new_stage)

sdk_response = ango_sdk.update_workflow_stages(project_id=project_id, stages=new_stages)

Prevent requeueing to Complete stage

python

import os
from dotenv import load_dotenv
from imerit_ango.sdk import SDK

load_dotenv('variables.env')
api_key = os.getenv('API_KEY')
project_id = os.getenv('PROJECT_ID')

ango_sdk = SDK(api_key)

stages = ango_sdk.get_project(project_id=project_id).get("data").get("project").get("stages")

new_stages = []
for stage in stages:
    new_stage = stage.copy()
    if stage["id"] == "Complete":
        new_stage["preventRequeue"] = True
    new_stages.append(new_stage)

sdk_response = ango_sdk.update_workflow_stages(project_id=project_id, stages=new_stages)

See also

get_project