Workflows and Stages

Workflow Basics

Encord SDK supports programmatic use of Workflows with the following: WorkflowStage, WorkflowAction, and Task. Typical use would be querying a WorkflowStage to get Tasks matching some criteria, and then executing a WorkflowAction against Tasks (for example: assigning, submitting annotation tasks, or approving review tasks). The type of actions available depends on the stage queried. Here is a short description of major parts of the Workflows SDK:

WorkflowStage

WorkflowStage: Move your tasks through a Workflow Project using the WorkflowStage class.
- AnnotationStage
- ReviewStage
- ConsensusAnnotationStage
- ConsensusReviewStage
- AgentStage
- FinalStage (Complete and Archive)

WorkflowAction

WorkflowAction: Applies an action to a task in a workflow stage The following actions are supported:
- assign: Assigns a user to one or more tasks.
- submit: Moves a task to the next stage.
- release: Releases a task from the current user.
- approve: Approves a review task.
- reject: Rejects a review task.
- move: Moves a task directly to any stage. See Move Tasks Freely.
- proceed: Moves an Agent task along one of the Agent stage’s pathways. See Move Agent tasks along a pathway.
Annotation, Review, Consensus Review, and Agent tasks also have an issues property for working with the task’s issues (for example, task.issues.add_file_issue(...)). See Issues & Comments in Projects.

Task

Task: Exposes a simple version of the tasks available in a Project.

TaskStatus

TaskStatus: The following statuses are available: The following statuses are available for ANNOTATE tasks:
- NEW: Tasks that are unassigned.
- ASSIGNED: Tasks that are assigned to a user.
- RELEASED: Tasks that were released from an assigned user.
- REOPENED: Tasks that were reopened.
- SKIPPED: Tasks that were skipped by annotators.
- COMPLETED: Tasks that are in the Complete stage. Used for Annotation sub-tasks in Consensus Annotation tasks.

Consensus Annotation tasks consist of 1 or more Annotation sub-tasks.

The following statuses are available for REVIEW and CONSENSUS REVIEW tasks:

NEW: Tasks that are unassigned.
ASSIGNED: Tasks that are assigned to a user.
RELEASED: Tasks that were released from an assigned user.
REOPENED: Tasks that were reopened.
SKIPPED: Tasks that were skipped by reviewers.

AGENT tasks use the same statuses as ANNOTATE tasks.All task status enums (AnnotationTaskStatus, ReviewTaskStatus, ConsensusReviewTaskStatus, and AgentTaskStatus), as well as LabelReviewStatus, also include:

UNKNOWN: Returned when the Encord platform reports a status that your version of the SDK does not recognize. Upgrade the encord SDK to see the actual status.

SKIPPED for Review and Consensus Review tasks, and the UNKNOWN fallback, require encord SDK version 0.1.207 or later.

In Consensus Projects, tasks in the COMPLETE and ARCHIVE blocks CANNOT be reopened.

View all stages in a Workflow

The following code provides you with the method to view all stages in a Workflow in a Project. The code is the same for Consensus and Non-Consensus Projects.

The SDK only returns stages that hold tasks: Annotate, Review, Consensus Annotate, Consensus Review, Agent, Complete, and Archive. Routers, Export, Metadata router, Parallelisation, and Checkpoint stages are not returned, so you cannot read their settings (for example, an Export stage’s destination) with the SDK.

# Import dependencies
from encord import EncordUserClient, Project


SSH_PATH = "<file-path-to-private-key-file>"
PROJECT_HASH = "<unique-id-for-project>"

user_client: EncordUserClient = EncordUserClient.create_with_ssh_private_key(
    ssh_private_key_path=SSH_PATH,
    # For US platform users use "https://api.us.encord.com"
    domain="https://api.encord.com",
)

# Get project
project: Project = user_client.get_project(PROJECT_HASH)

# Print all Workflow stages
for idx, stage in enumerate(project.workflow.stages):
    print(f"Stage #{idx}: {stage}")

# Import dependencies
from encord import EncordUserClient, Project


SSH_PATH = "/Users/chris-encord/sdk-ssh-private-key.txt"
PROJECT_HASH = "2c2a74d0-d33d-43b2-b5dd-1c6c3e6d7b6d"

user_client: EncordUserClient = EncordUserClient.create_with_ssh_private_key(
    ssh_private_key_path=SSH_PATH,
    # For US platform users use "https://api.us.encord.com"
    domain="https://api.encord.com",
)

# Get project
project: Project = user_client.get_project(PROJECT_HASH)

# Print all Workflow stages
for idx, stage in enumerate(project.workflow.stages):
    print(f"Stage #{idx}: {stage}")

Stage #0: AnnotationStage(stage_type=WorkflowStageType.ANNOTATION, uuid='2ed1d9f4-5b42-4959-bcbe-93eaf866320c', title='Annotate 1')
Stage #1: ReviewStage(stage_type=WorkflowStageType.REVIEW, uuid='ddb7973d-bae9-4ced-8dcb-ec178e66dba0', title='Review 1')
Stage #2: AnnotationStage(stage_type=WorkflowStageType.ANNOTATION, uuid='58ec51ba-2669-4036-b25c-17aecf805a59', title='Annotate 2')
Stage #3: ReviewStage(stage_type=WorkflowStageType.REVIEW, uuid='8047e0d5-fc41-4ac5-a7ed-335b3dbd887f', title='Review 2')
Stage #4: FinalStage(stage_type=WorkflowStageType.DONE, uuid='90fc8b51-6a2f-4de0-88d0-059f76865385', title='Complete')
Stage #5: FinalStage(stage_type=WorkflowStageType.DONE, uuid='1432dbf0-5c19-41fe-a7b0-811d36b27456', title='Archive')

Stage #0: ConsensusAnnotationStage(stage_type=WorkflowStageType.CONSENSUS_ANNOTATION, uuid='2adfe595-4a56-41f2-bfcc-eeab3dd58b65', title='Consensus 1')
Stage #1: ConsensusReviewStage(stage_type=WorkflowStageType.CONSENSUS_REVIEW, uuid='1f37fb4c-a74a-4437-b33d-6fc4e3ccc2f5', title='Consensus 1 Review')
Stage #2: ReviewStage(stage_type=WorkflowStageType.REVIEW, uuid='9fc321da-0e62-4cc5-b5e7-5ce8f73c3c20', title='Review 1')
Stage #3: ConsensusAnnotationStage(stage_type=WorkflowStageType.CONSENSUS_ANNOTATION, uuid='518cd79f-a9a9-4ab9-bb92-3dcf0fae29a2', title='Consensus 2')
Stage #4: ConsensusReviewStage(stage_type=WorkflowStageType.CONSENSUS_REVIEW, uuid='81e0cb65-69ae-4d3b-aac8-8917f1480930', title='Consensus 2 Review')
Stage #5: FinalStage(stage_type=WorkflowStageType.DONE, uuid='fd5c3513-c7a5-42c3-a07c-933838aecf62', title='Complete')
Stage #6: FinalStage(stage_type=WorkflowStageType.DONE, uuid='c67d08c5-b93b-49be-87b2-39809ea7b27a', title='Archive')

Non-Consensus Project This is an example of a non-Consensus Workflow graph.

Consensus Workflow Project This is an example of a Consensus Workflow graph.

Non-Consensus Projects

In the SDK, a class is defined for every stage type. This allows you to see what properties and actions each stage offers. The following stages are supported for non-Consensus Projects:

AnnotationStage
ReviewStage
AgentStage
FinalStage (Complete and Archive)

Every stage has a uuid and a title, and a get_tasks() method that returns the stage’s tasks. You can filter tasks with these optional arguments:

assignee: A user email, or a list of user emails.
data_hash: A data unit ID, or a list of data unit IDs.
dataset_hash: A dataset ID, or a list of dataset IDs.
data_title: Text that the data unit’s title contains.
status: A task status, or a list of task statuses (for example, AnnotationTaskStatus.ASSIGNED). Not available for the Complete and Archive stages, which also don’t take assignee.

Tasks provide programmatic access to the tasks available in the UI. The following actions are supported for non-Consensus Projects:

assign: Assigns a user to one or more tasks.
submit: Moves a task to the next stage.
release: Releases a task from the current user.
approve and reject: Approve or reject a review task.
move: Moves a task directly to any stage.

An annotation task has the following fields:

class AnnotationTask:
    uuid: UUID  # the task ID
    created_at: datetime
    updated_at: datetime
    status: AnnotationTaskStatus
    data_hash: UUID  # the data unit ID
    data_title: str
    label_branch_name: str  # "main" by default
    assignee: Optional[str]  # email of the assigned user

Display Tasks in a Stage

Use the following scripts as templates for annotation and review tasks.

Display Stage Tasks

# Import dependencies
from encord import EncordUserClient, Project

from encord.workflow import(
  AnnotationStage, 
  ReviewStage,
  ConsensusAnnotationStage, 
  ConsensusReviewStage,
  FinalStage
)

SSH_PATH = "/Users/chris-encord/sdk-ssh-private-key.txt"
PROJECT_HASH = "<project-unique-id>"

user_client: EncordUserClient = EncordUserClient.create_with_ssh_private_key(
    ssh_private_key_path=SSH_PATH,
    # For US platform users use "https://api.us.encord.com"
    domain="https://api.encord.com",
)

# Get project
project: Project = user_client.get_project(PROJECT_HASH)

stage = project.workflow.get_stage(name="Annotate 1", type_=AnnotationStage)
# type_ is an optional parameter here, any other type narrowing technique can be used here
# for example it can be an assert as follows
assert isinstance(stage, AnnotationStage)

for task in stage.get_tasks():
    print(f"Task: {task}")

Assign a User to Tasks in a Stage

Use the following scripts as templates for annotation and review tasks.

from pathlib import Path
from encord.user_client import EncordUserClient
from encord.workflow.stages.annotation import AnnotationStage, AnnotationTaskStatus
from encord.workflow.stages.consensus_annotation import ConsensusAnnotationStage
from encord.workflow.stages.consensus_review import ConsensusReviewStage
from encord.workflow.stages.review import ReviewStage

SSH_PATH = "<file-path-to-private-key>"

PROJECT_ID="<project-unique-id>"

user_client: EncordUserClient = EncordUserClient.create_with_ssh_private_key(
    Path(SSH_PATH).read_text(),
    # For US platform users use "https://api.us.encord.com"
    domain="https://api.encord.com",
)

project = user_client.get_project(PROJECT_ID)

workflow = project.workflow

annotation_stage = workflow.get_stage(name="<name-of-workflow-stage>", type_=<type-of-workflow-stage>)

BUNDLE_SIZE = 100  # You can adjust this value as needed, but keep it <= 1000

with project.create_bundle(bundle_size=BUNDLE_SIZE) as bundle:
    for annotation_task in annotation_stage.get_tasks():
        print(annotation_task.data_title)
        
        # Assign task to someone
        annotation_task.assign(assignee='<email-address>', bundle=bundle)

from pathlib import Path
from encord.user_client import EncordUserClient
from encord.workflow.stages.annotation import AnnotationStage, AnnotationTaskStatus
from encord.workflow.stages.consensus_annotation import ConsensusAnnotationStage
from encord.workflow.stages.consensus_review import ConsensusReviewStage
from encord.workflow.stages.review import ReviewStage

SSH_PATH = "/Users/chris-encord/sdk-ssh-private-key.txt"

PROJECT_ID="<project-unique-id>"

user_client: EncordUserClient = EncordUserClient.create_with_ssh_private_key(
    Path(SSH_PATH).read_text(),
    # For US platform users use "https://api.us.encord.com"
    domain="https://api.encord.com",
)

project = user_client.get_project(PROJECT_ID)

workflow = project.workflow

annotation_stage = workflow.get_stage(name="Annotate 1", type_=AnnotationStage)

BUNDLE_SIZE = 100  # You can adjust this value as needed, but keep it <= 1000

with project.create_bundle(bundle_size=BUNDLE_SIZE) as bundle:
    for annotation_task in annotation_stage.get_tasks():
        print(annotation_task.data_title)
        
        # Assign task to someone
        annotation_task.assign(assignee='chris@acme.com', bundle=bundle)

Submit Tasks as Another User in a Stage

Annotation Tasks can only be submitted on behalf of users with the following roles: Annotator, Annotator + Reviewer, Team Manager, or Admin.
Review Tasks can only be submitted on behalf of users with the following roles: Reviewer, Annotator + Reviewer, Team Manager, or Admin.

from pathlib import Path
from encord.user_client import EncordUserClient
from encord.workflow.stages.annotation import AnnotationStage, AnnotationTaskStatus
from encord.workflow.stages.consensus_annotation import ConsensusAnnotationStage
from encord.workflow.stages.consensus_review import ConsensusReviewStage
from encord.workflow.stages.review import ReviewStage

SSH_PATH = "<file-path-to-private-key>"

PROJECT_ID="<project-unique-id>"

user_client: EncordUserClient = EncordUserClient.create_with_ssh_private_key(
    Path(SSH_PATH).read_text(),
    # For US platform users use "https://api.us.encord.com"
    domain="https://api.encord.com",
)

project = user_client.get_project(PROJECT_ID)

workflow = project.workflow

annotation_stage = workflow.get_stage(name="<name-of-annotation-stage>", type_=<type-of-workflow-stage>)

BUNDLE_SIZE = 100  # You can adjust this value as needed, but keep it <= 1000

with project.create_bundle(bundle_size=BUNDLE_SIZE) as bundle:
    for annotation_task in annotation_stage.get_tasks():
        print(annotation_task.data_title)
        
        # Submit as someone explicitly
        annotation_task.submit(assignee='<email-address>', bundle=bundle)

from pathlib import Path
from encord.user_client import EncordUserClient
from encord.workflow.stages.annotation import AnnotationStage, AnnotationTaskStatus
from encord.workflow.stages.consensus_annotation import ConsensusAnnotationStage
from encord.workflow.stages.consensus_review import ConsensusReviewStage
from encord.workflow.stages.review import ReviewStage

SSH_PATH = "<file-path-to-private-key>"

PROJECT_ID="<project-unique-id>"

user_client: EncordUserClient = EncordUserClient.create_with_ssh_private_key(
    Path(SSH_PATH).read_text(),
    # For US platform users use "https://api.us.encord.com"
    domain="https://api.encord.com",
)

project = user_client.get_project(PROJECT_ID)

workflow = project.workflow

review_stage = workflow.get_stage(name="<name-of-review-stage>", type_=<type-of-workflow-stage>)

BUNDLE_SIZE = 100  # You can adjust this value as needed, but keep it <= 1000

with project.create_bundle(bundle_size=BUNDLE_SIZE) as bundle:
    for review_task in review_stage.get_tasks():
        print(review_task.data_title)
        
        # Submit as someone explicitly
        review_task.submit(assignee='<email-address>', bundle=bundle)

from pathlib import Path
from encord.user_client import EncordUserClient
from encord.workflow.stages.annotation import AnnotationStage, AnnotationTaskStatus
from encord.workflow.stages.consensus_annotation import ConsensusAnnotationStage
from encord.workflow.stages.consensus_review import ConsensusReviewStage
from encord.workflow.stages.review import ReviewStage

SSH_PATH = "/Users/chris-encord/sdk-ssh-private-key.txt"

PROJECT_ID="<project-unique-id>"

user_client: EncordUserClient = EncordUserClient.create_with_ssh_private_key(
    Path(SSH_PATH).read_text(),
    # For US platform users use "https://api.us.encord.com"
    domain="https://api.encord.com",
)

project = user_client.get_project(PROJECT_ID)

workflow = project.workflow

annotation_stage = workflow.get_stage(name="Annotate 1", type_=AnnotationStage)

BUNDLE_SIZE = 100  # You can adjust this value as needed, but keep it <= 1000

with project.create_bundle(bundle_size=BUNDLE_SIZE) as bundle:
    for annotation_task in annotation_stage.get_tasks():
        print(annotation_task.data_title)
        
        # Submit as someone explicitly
        annotation_task.submit(assignee='chris@acme.com', bundle=bundle)

Retain Current User and Submit Task

Annotation Tasks can only be submitted by users with the following roles: Annotator, Annotator + Reviewer, Team Manager, or Admin.
Review Tasks can only be submitted by users with the following roles: Reviewer, Annotator + Reviewer, Team Manager, or Admin.

from pathlib import Path
from encord.user_client import EncordUserClient
from encord.workflow.stages.annotation import AnnotationStage, AnnotationTaskStatus
from encord.workflow.stages.consensus_annotation import ConsensusAnnotationStage
from encord.workflow.stages.consensus_review import ConsensusReviewStage
from encord.workflow.stages.review import ReviewStage

SSH_PATH = "<file-path-to-private-key>"

PROJECT_ID="<project-unique-id>"

user_client: EncordUserClient = EncordUserClient.create_with_ssh_private_key(
    Path(SSH_PATH).read_text(),
    # For US platform users use "https://api.us.encord.com"
    domain="https://api.encord.com",
)

project = user_client.get_project(PROJECT_ID)

workflow = project.workflow

annotation_stage = workflow.get_stage(name="<name-of-workflow-stage>", type_=<type-of-workflow-stage>)

BUNDLE_SIZE = 100  # You can adjust this value as needed, but keep it <= 1000

with project.create_bundle(bundle_size=BUNDLE_SIZE) as bundle:
    for annotation_task in annotation_stage.get_tasks():
        print(annotation_task.data_title)
        
        # Retain current assignee and submit
        annotation_task.submit(retain_assignee=True, bundle=bundle)

from pathlib import Path
from encord.user_client import EncordUserClient
from encord.workflow.stages.annotation import AnnotationStage, AnnotationTaskStatus
from encord.workflow.stages.consensus_annotation import ConsensusAnnotationStage
from encord.workflow.stages.consensus_review import ConsensusReviewStage
from encord.workflow.stages.review import ReviewStage

SSH_PATH = "<file-path-to-private-key>"

PROJECT_ID="<project-unique-id>"

user_client: EncordUserClient = EncordUserClient.create_with_ssh_private_key(
    Path(SSH_PATH).read_text(),
    # For US platform users use "https://api.us.encord.com"
    domain="https://api.encord.com",
)

project = user_client.get_project(PROJECT_ID)

workflow = project.workflow

review_stage = workflow.get_stage(name="<name-of-review-stage>", type_=<type-of-workflow-stage>)

BUNDLE_SIZE = 100  # You can adjust this value as needed, but keep it <= 1000

with project.create_bundle(bundle_size=BUNDLE_SIZE) as bundle:
    for review_task in review_stage.get_tasks():
        print(review_task.data_title)
        
        # Retain current assignee and submit
        review_task.submit(retain_assignee=True, bundle=bundle)

from pathlib import Path
from encord.user_client import EncordUserClient
from encord.workflow.stages.annotation import AnnotationStage, AnnotationTaskStatus
from encord.workflow.stages.consensus_annotation import ConsensusAnnotationStage
from encord.workflow.stages.consensus_review import ConsensusReviewStage
from encord.workflow.stages.review import ReviewStage

SSH_PATH = "/Users/chris-encord/sdk-ssh-private-key.txt"

PROJECT_ID="<project-unique-id>"

user_client: EncordUserClient = EncordUserClient.create_with_ssh_private_key(
    Path(SSH_PATH).read_text(),
    # For US platform users use "https://api.us.encord.com"
    domain="https://api.encord.com",
)

project = user_client.get_project(PROJECT_ID)

workflow = project.workflow

annotation_stage = workflow.get_stage(name="Annotate 1", type_=AnnotationStage)

BUNDLE_SIZE = 100  # You can adjust this value as needed, but keep it <= 1000

with project.create_bundle(bundle_size=BUNDLE_SIZE) as bundle:
    for annotation_task in annotation_stage.get_tasks():
        print(annotation_task.data_title)
        
        # Retain current assignee and submit
        annotation_task.submit(retain_assignee=True, bundle=bundle)

Release Tasks in a Stage

Releasing a task removes its current assignee and returns the task to the stage’s queue, so any eligible user can pick it up. Use release() on annotation tasks, review tasks, and Consensus review tasks.

# Import dependencies
from encord import EncordUserClient
from encord.workflow.stages.annotation import AnnotationStage, AnnotationTaskStatus

SSH_PATH = "<file-path-to-private-key>"
PROJECT_ID = "<project-unique-id>"
ASSIGNEE_EMAIL = "<email-of-assigned-user>"

user_client: EncordUserClient = EncordUserClient.create_with_ssh_private_key(
    ssh_private_key_path=SSH_PATH,
    # For US platform users use "https://api.us.encord.com"
    domain="https://api.encord.com",
)

project = user_client.get_project(PROJECT_ID)

annotation_stage = project.workflow.get_stage(name="<name-of-annotation-stage>", type_=AnnotationStage)

BUNDLE_SIZE = 100  # You can adjust this value as needed, but keep it <= 1000

# Release all tasks currently assigned to a specific user
with project.create_bundle(bundle_size=BUNDLE_SIZE) as bundle:
    for annotation_task in annotation_stage.get_tasks(
        assignee=ASSIGNEE_EMAIL, status=AnnotationTaskStatus.ASSIGNED
    ):
        print(f"Releasing {annotation_task.data_title}")
        annotation_task.release(bundle=bundle)

# Import dependencies
from encord import EncordUserClient
from encord.workflow.stages.review import ReviewStage, ReviewTaskStatus

SSH_PATH = "<file-path-to-private-key>"
PROJECT_ID = "<project-unique-id>"
ASSIGNEE_EMAIL = "<email-of-assigned-user>"

user_client: EncordUserClient = EncordUserClient.create_with_ssh_private_key(
    ssh_private_key_path=SSH_PATH,
    # For US platform users use "https://api.us.encord.com"
    domain="https://api.encord.com",
)

project = user_client.get_project(PROJECT_ID)

review_stage = project.workflow.get_stage(name="<name-of-review-stage>", type_=ReviewStage)

BUNDLE_SIZE = 100  # You can adjust this value as needed, but keep it <= 1000

# Release all review tasks currently assigned to a specific user
with project.create_bundle(bundle_size=BUNDLE_SIZE) as bundle:
    for review_task in review_stage.get_tasks(assignee=ASSIGNEE_EMAIL, status=ReviewTaskStatus.ASSIGNED):
        print(f"Releasing {review_task.data_title}")
        review_task.release(bundle=bundle)

# Import dependencies
from encord import EncordUserClient
from encord.workflow.stages.consensus_review import ConsensusReviewStage

SSH_PATH = "<file-path-to-private-key>"
PROJECT_ID = "<project-unique-id>"
ASSIGNEE_EMAIL = "<email-of-assigned-user>"

user_client: EncordUserClient = EncordUserClient.create_with_ssh_private_key(
    ssh_private_key_path=SSH_PATH,
    # For US platform users use "https://api.us.encord.com"
    domain="https://api.encord.com",
)

project = user_client.get_project(PROJECT_ID)

consensus_review_stage = project.workflow.get_stage(
    name="<name-of-consensus-review-stage>", type_=ConsensusReviewStage
)

BUNDLE_SIZE = 100  # You can adjust this value as needed, but keep it <= 1000

# Release all Consensus review tasks currently assigned to a specific user
with project.create_bundle(bundle_size=BUNDLE_SIZE) as bundle:
    for review_task in consensus_review_stage.get_tasks(assignee=ASSIGNEE_EMAIL):
        print(f"Releasing {review_task.data_title}")
        review_task.release(bundle=bundle)

Reopen Label Reviews in a Review Task

Each label in a review task has its own label review, which can be approved, rejected, or reopened. Use get_label_reviews() to list the label reviews for a task, and reopen() to set an approved or rejected label review back to an open state so that it can be reviewed again.

Reopen Label Reviews

# Import dependencies
from encord import EncordUserClient
from encord.workflow.stages.review import LabelReviewStatus, ReviewStage

SSH_PATH = "<file-path-to-private-key>"
PROJECT_ID = "<project-unique-id>"
DATA_TITLE = "<name-of-data-unit>"

user_client: EncordUserClient = EncordUserClient.create_with_ssh_private_key(
    ssh_private_key_path=SSH_PATH,
    # For US platform users use "https://api.us.encord.com"
    domain="https://api.encord.com",
)

project = user_client.get_project(PROJECT_ID)

review_stage = project.workflow.get_stage(name="<name-of-review-stage>", type_=ReviewStage)

for review_task in review_stage.get_tasks(data_title=DATA_TITLE):
    for label_review in review_task.get_label_reviews():
        print(f"{label_review.label_type} {label_review.label_id}: {label_review.status}")

        # Reopen label reviews that were rejected
        if label_review.status == LabelReviewStatus.REJECTED:
            label_review.reopen()

Get all Tasks

Get Tasks

# Import dependencies
from encord import EncordUserClient, Project

from encord.workflow import(
  AnnotationStage, 
  ReviewStage,
  ConsensusAnnotationStage, 
  ConsensusReviewStage,
  FinalStage
)

SSH_PATH = "/Users/chris-encord/sdk-ssh-private-key.txt"
PROJECT_HASH = "<project-unique-id>"

user_client: EncordUserClient = EncordUserClient.create_with_ssh_private_key(
    ssh_private_key_path=SSH_PATH,
    # For US platform users use "https://api.us.encord.com"
    domain="https://api.encord.com",
)

# Get project
project: Project = user_client.get_project(PROJECT_HASH)

stage = project.workflow.get_stage(name="Consensus 1", type_=ConsensusAnnotationStage)
# type_ is an optional parameter here, any other type narrowing technique can be used here
# for example it can be an assert like follows
assert isinstance(stage, ConsensusAnnotationStage)

for task in stage.get_tasks():
    
    print(f"Task: {task}")

Filter tasks by data hash

You can filter tasks in any stage using data hashes.

Filter tasks by data hash

# Import dependencies
from encord import EncordUserClient, Project
from encord.workflow import AnnotationStage, ReviewStage, ConsensusAnnotationStage, ConsensusReviewStage, FinalStage

# Constants
SSH_PATH = "<ssh-private-key>"
PROJECT_HASH = "<project-unique-id>"

# Create user client
user_client: EncordUserClient = EncordUserClient.create_with_ssh_private_key(
    ssh_private_key_path=SSH_PATH,
    # For US platform users use "https://api.us.encord.com"
    domain="https://api.encord.com",
)

# Get project
project: Project = user_client.get_project(PROJECT_HASH)

# Get stage
stage = project.workflow.get_stage(name="Review 1", type_=ReviewStage)
# Uncomment the following line for different stage
# stage = project.workflow.get_stage(name="Annotate 1", type_=AnnotationStage)

# Check instance type. Change the type for each stage
assert isinstance(stage, ReviewStage)

# List of data hashes.
data_hashes = ["<data-unique-id-01>", "<data-unique-id-02>", "<data-unique-id-03>"]

for task in stage.get_tasks(data_hash=data_hashes):
    print(f"Task: {task}")

Filter tasks by status

Use the status parameter of get_tasks() to return only tasks with one or more specific statuses. Each stage type has its own status enum: AnnotationTaskStatus, ReviewTaskStatus, ConsensusReviewTaskStatus, or AgentTaskStatus.If the Encord platform returns a status that your SDK version does not recognize, the task’s status is UNKNOWN rather than raising an error.

Filter tasks by status

# Import dependencies
from encord import EncordUserClient, Project
from encord.workflow.stages.review import ReviewStage, ReviewTaskStatus

# Constants
SSH_PATH = "<ssh-private-key>"
PROJECT_HASH = "<project-unique-id>"

# Create user client
user_client: EncordUserClient = EncordUserClient.create_with_ssh_private_key(
    ssh_private_key_path=SSH_PATH,
    # For US platform users use "https://api.us.encord.com"
    domain="https://api.encord.com",
)

# Get project
project: Project = user_client.get_project(PROJECT_HASH)

# Get stage
stage = project.workflow.get_stage(name="Review 1", type_=ReviewStage)

# Get only skipped and reopened tasks
for task in stage.get_tasks(status=[ReviewTaskStatus.SKIPPED, ReviewTaskStatus.REOPENED]):
    print(f"Task: {task.data_title}, status: {task.status}")

# Handle statuses that are newer than your SDK version
for task in stage.get_tasks():
    if task.status == ReviewTaskStatus.UNKNOWN:
        print(f"Task {task.data_title} has a status this SDK version does not recognize. Upgrade the encord SDK.")

Move all tasks in a stage to the next stage

You can submit tasks as the currently assigned user or as a different user. If you do not specify the user to submit the task, you are assigned as the user submitting the task.

# Import dependencies
from encord import EncordUserClient, Project
from encord.workflow import AnnotationStage

SSH_PATH = "<file-path-to-private-key>"
PROJECT_HASH = "<project-unique-id>"

# Authenticate
user_client: EncordUserClient = EncordUserClient.create_with_ssh_private_key(
    ssh_private_key_path=SSH_PATH,
    # For US platform users use "https://api.us.encord.com"
    domain="https://api.encord.com",
)

# Get project
project: Project = user_client.get_project(PROJECT_HASH)

# Get the specific stage (in this case, "Annotate 1")
stage = project.workflow.get_stage(name="Annotate 1", type_=AnnotationStage)
assert isinstance(stage, AnnotationStage)

# Define bundle size
BUNDLE_SIZE = 100  # You can adjust this value as needed, but keep it <= 1000

# Create a bundle and move tasks
with project.create_bundle(bundle_size=BUNDLE_SIZE) as bundle:
    for task in stage.get_tasks():
        # The task is submitted as the user who is currently assigned to the task
        task.submit(retain_assignee=True, bundle=bundle)
        print(f"Task: {task}")

print("All tasks have been processed and moved to the next stage.")

# Import dependencies
from encord import EncordUserClient, Project
from encord.workflow import AnnotationStage

SSH_PATH = "<file-path-to-private-key>"
PROJECT_HASH = "<project-unique-id>"

# Authenticate
user_client: EncordUserClient = EncordUserClient.create_with_ssh_private_key(
    ssh_private_key_path=SSH_PATH,
    # For US platform users use "https://api.us.encord.com"
    domain="https://api.encord.com",
)

# Get project
project: Project = user_client.get_project(PROJECT_HASH)

# Get the specific stage (in this case, "Annotate 1")
stage = project.workflow.get_stage(name="Annotate 1", type_=AnnotationStage)
assert isinstance(stage, AnnotationStage)

# Define bundle size
BUNDLE_SIZE = 100  # You can adjust this value as needed, but keep it <= 1000

# Create a bundle and move tasks
with project.create_bundle(bundle_size=BUNDLE_SIZE) as bundle:
    for task in stage.get_tasks():
        # The task is submitted as the user you specify.
        task.submit(assignee='<email-of-user-you-want-as-assignee>', bundle=bundle)
        print(f"Task: {task}")

print("All tasks have been processed and moved to the next stage.")

Move Agent tasks along a pathway

An Agent stage has one or more pathways, each leading to another stage. Use proceed() to send a task along a pathway, by its name or its UUID. To run agents that do this automatically, see Task Agents.

# Import dependencies
from encord import EncordUserClient
from encord.workflow import AgentStage

SSH_PATH = "/Users/chris-encord/sdk-ssh-private-key.txt"
PROJECT_HASH = "<project-unique-id>"

user_client: EncordUserClient = EncordUserClient.create_with_ssh_private_key(
    ssh_private_key_path=SSH_PATH,
    # For US platform users use "https://api.us.encord.com"
    domain="https://api.encord.com",
)

project = user_client.get_project(PROJECT_HASH)

agent_stage = project.workflow.get_stage(name="Agent 1", type_=AgentStage)

# List the pathways out of the Agent stage
for pathway in agent_stage.pathways:
    print(f"Pathway: {pathway.name} ({pathway.uuid})")

# Send each task along a pathway
with project.create_bundle() as bundle:
    for task in agent_stage.get_tasks():
        # Replace with your own logic to choose a pathway for each task
        task.proceed(pathway_name="<pathway-name>", bundle=bundle)

Use model predictions as an annotator

# Import dependencies
from encord import EncordUserClient, Project

from encord.workflow import (
    AnnotationStage,
    ReviewStage,
    ConsensusAnnotationStage,
    ConsensusReviewStage,
    FinalStage,
)

SSH_PATH = "/Users/chris-encord/sdk-ssh-private-key.txt"
PROJECT_HASH = "<project-unique-id>"

user_client: EncordUserClient = EncordUserClient.create_with_ssh_private_key(
    ssh_private_key_path=SSH_PATH,
    # For US platform users use "https://api.us.encord.com"
    domain="https://api.encord.com",
)

# Get project
project: Project = user_client.get_project(PROJECT_HASH)

stage = project.workflow.get_stage(name="Automated Annotation", type_=AnnotationStage)
assert isinstance(stage, AnnotationStage)

BUNDLE_SIZE = 100  # Define bundle size

# List the tasks once, and keep the ones you want to pre-annotate
tasks = [task for task in stage.get_tasks() if match_some_criteria(task)]

# Assign tasks in bundles
with project.create_bundle(bundle_size=BUNDLE_SIZE) as bundle:
    for task in tasks:
        task.assign("admin-alex@encord.com", bundle=bundle)

# Get the label rows for all the tasks in one call, and initialize them in bundles
label_rows = project.list_label_rows_v2(data_hashes=[task.data_hash for task in tasks])
with project.create_bundle(bundle_size=BUNDLE_SIZE) as bundle:
    for label_row in label_rows:
        label_row.initialise_labels(bundle=bundle)

# Custom logic that populates label rows with labels.
# This runs after the bundle above has finished, so the labels are initialized.
for label_row in label_rows:
    do_preannotate(label_row)

# Save label rows in bundles. Save before you submit: submitting a task does not upload labels.
with project.create_bundle(bundle_size=BUNDLE_SIZE) as bundle:
    for label_row in label_rows:
        label_row.save(bundle=bundle)

# Submit tasks in bundles
with project.create_bundle(bundle_size=BUNDLE_SIZE) as bundle:
    for task in tasks:
        task.submit(bundle=bundle)
        print(f"Task submitted: {task.data_title}")

Consensus Projects

Consensus annotation tasks can be listed, but cannot be assigned or moved into the Consensus review stage. In Consensus Projects, tasks are broken into subtasks. One subtask for each Annotator. Subtasks can be assigned or submitted. Consensus Projects support the following stages:

AnnotationStage
ReviewStage
ConsensusAnnotationStage
ConsensusReviewStage
FinalStage (Complete and Archive)

The following actions are supported for Consensus review and Review and Refine:

assign: Assigns a user to one or more tasks.
approve: Moves a task to the next stage.
release: Releases a task from the current user.
reject: Rejects a task.

Get all Subtasks

Get Subtasks

# Import dependencies
from encord import EncordUserClient, Project

from encord.workflow import(
  AnnotationStage, 
  ReviewStage,
  ConsensusAnnotationStage, 
  ConsensusReviewStage,
  FinalStage
)

SSH_PATH = "/Users/chris-encord/sdk-ssh-private-key.txt"
PROJECT_HASH = "<project-unique-id>"

user_client: EncordUserClient = EncordUserClient.create_with_ssh_private_key(
    ssh_private_key_path=SSH_PATH,
    # For US platform users use "https://api.us.encord.com"
    domain="https://api.encord.com",
)

# Get project
project: Project = user_client.get_project(PROJECT_HASH)

stage = project.workflow.get_stage(name="Consensus 1", type_=ConsensusAnnotationStage)
# type_ is an optional parameter here, any other type narrowing technique can be used here
# for example it can be an assert like follows
assert isinstance(stage, ConsensusAnnotationStage)

for task in stage.get_tasks():
    
    print(f"Task: {task}")
    for subtask in task.subtasks:
        print(f"   subtask: {subtask}")

Select your labels from a Consensus Project

The following code includes produce_result, that is undefined. You need to create and define the custom logic that selects the labels from the label options you have available.

# Import dependencies
from encord import EncordUserClient, Project

from encord.workflow import(
  AnnotationStage, 
  ReviewStage,
  ConsensusAnnotationStage, 
  ConsensusReviewStage,
  FinalStage
)

SSH_PATH = "/Users/chris-encord/sdk-ssh-private-key.txt"
PROJECT_HASH = "<project-unique-id>"

user_client: EncordUserClient = EncordUserClient.create_with_ssh_private_key(
    ssh_private_key_path=SSH_PATH,
    # For US platform users use "https://api.us.encord.com"
    domain="https://api.encord.com",
)

# Get project
project: Project = user_client.get_project(PROJECT_HASH)

# Obtaining the stage
stage = project.workflow.get_stage(name="Consensus 1 Review", type_=ConsensusReviewStage)

BUNDLE_SIZE = 100  # Define bundle size

# Get all the tasks (or rather iterator over tasks), 
# Do NOT download them all at once.
for task in stage.get_tasks():
    # Each consensus branch has a unique label hash, so use them as filter
    label_hashes = [option.label_hash for option in task.options]
    label_rows_for_consensus = project.list_label_rows_v2(label_hashes=label_hashes)

    target_label_row = project.list_label_rows_v2(data_hashes=[task.data_hash])[0]
   
    with project.create_bundle(bundle_size=BUNDLE_SIZE) as label_initialize_bundle:
        # Bundling all our initialization operations in one operation
        for label_row in label_rows_for_consensus:
            label_row.initialise_labels(bundle=label_initialize_bundle)

        target_label_row.initialise_labels(bundle=label_initialize_bundle)

    # Perform custom user logic that compares the consensus label rows and populates
    # the target label row. produce_result is not defined in the Encord SDK: you need
    # to define it in your own code.
    is_consensus = produce_result(label_rows_for_consensus, target_label_row)

    # Approve or reject each task once, after comparing all of its label rows
    if is_consensus:
        target_label_row.save()
        task.assign("<your-admin@example.com>")
        task.approve()
    else:
        task.reject()

Work with Labels on Tasks

A task tracks where a data unit is in the Workflow. The data unit’s labels are in its label row. Use the task’s data_hash to get the label row.

Save labels before you submit a task. submit(), approve(), and the other task actions only move the task. They don’t upload labels. Call label_row.save() first, or labels you added in your script are not included.
You can edit labels on tasks in the Complete stage. Saving labels does not move the task: it stays in the Complete stage, so you don’t need to reopen it first.
Saving labels on a task in a Review stage updates the task’s label reviews. New labels get a review, and the reviews of removed labels are removed. If no label reviews remain, the review task closes automatically.

The following code gets the label rows for all tasks in a stage in one call. Use data_hash to identify the data unit in the Project, and backing_item_uuid to identify the storage item, for example to add it to a Collection.

Get the label rows for tasks

# Import dependencies
from encord import EncordUserClient
from encord.workflow import FinalStage

SSH_PATH = "/Users/chris-encord/sdk-ssh-private-key.txt"
PROJECT_HASH = "<project-unique-id>"

user_client: EncordUserClient = EncordUserClient.create_with_ssh_private_key(
    ssh_private_key_path=SSH_PATH,
    # For US platform users use "https://api.us.encord.com"
    domain="https://api.encord.com",
)

project = user_client.get_project(PROJECT_HASH)

stage = project.workflow.get_stage(name="Complete", type_=FinalStage)

# List the tasks once, then get all their label rows in one call
tasks = list(stage.get_tasks())
label_rows = project.list_label_rows_v2(data_hashes=[task.data_hash for task in tasks])

for label_row in label_rows:
    print(f"{label_row.data_title}: data unit {label_row.data_hash}, storage item {label_row.backing_item_uuid}")

Move Tasks

To Next Stage

Submitting a task moves it to the next stage in the Workflow.

Submit Task To Next Stage

from pathlib import Path
from encord.user_client import EncordUserClient
from encord.workflow.stages.annotation import AnnotationStage, AnnotationTaskStatus
from encord.workflow.stages.consensus_annotation import ConsensusAnnotationStage
from encord.workflow.stages.consensus_review import ConsensusReviewStage
from encord.workflow.stages.review import ReviewStage

SSH_PATH = "/Users/chris-encord/sdk-ssh-private-key.txt"

PROJECT_ID="<project-unique-id>"

user_client: EncordUserClient = EncordUserClient.create_with_ssh_private_key(
    Path(SSH_PATH).read_text(),
    # For US platform users use "https://api.us.encord.com"
    domain="https://api.encord.com",
)

project = user_client.get_project(PROJECT_ID)

workflow = project.workflow

annotation_stage = workflow.get_stage(name="Annotate 1", type_=AnnotationStage)

BUNDLE_SIZE = 100  # You can adjust this value as needed, but keep it <= 1000

with project.create_bundle(bundle_size=BUNDLE_SIZE) as bundle:
    for annotation_task in annotation_stage.get_tasks():
        print(annotation_task.data_title)
        
        # Submit as someone explicitly
        annotation_task.submit(assignee='chris@acme.com', bundle=bundle)

To Complete

workflow_complete() moves Workflow tasks to the Complete stage.

In Consensus Projects, tasks in the COMPLETE and ARCHIVE blocks CANNOT be reopened.

# Import dependencies
from encord.user_client import EncordUserClient

#Authenticate using the path to your private key
user_client = EncordUserClient.create_with_ssh_private_key(
    ssh_private_key_path="<private_key_path>",
    # For US platform users use "https://api.us.encord.com"
    domain="https://api.encord.com",
    )

# Open the project you want to work on by specifying the project hash
project = user_client.get_project(
    "<project-hash>"
)

# Create a bundle
with project.create_bundle() as bundle:
    # Move all tasks into the final stage of the Workflow
    for label_row in project.list_label_rows_v2():
        label_row.workflow_complete(bundle=bundle)

# Import dependencies
from encord.user_client import EncordUserClient

#Authenticate using the path to your private key
user_client = EncordUserClient.create_with_ssh_private_key(
    ssh_private_key_path='<private_key_path>',
    # For US platform users use "https://api.us.encord.com"
    domain="https://api.encord.com",
)

# Open the project you want to work on by specifying the project hash
project = user_client.get_project(
    "<project-hash>"
)

# Move all tasks from <review_stage_2> into the final stage of the Workflow project
for label_row in project.list_label_rows_v2():
    if label_row.workflow_graph_node.title == "<review_stage_2>":
        label_row.workflow_complete()

Reopen

Use the workflow_reopen() method to reopen tasks. Reopening a task moves it back to the first annotation stage in your Workflow. No labels are lost.

workflow_reopen() raises an error in Projects whose first annotation stage is a Consensus stage, for every task in the Project.
workflow_reopen() and workflow_complete() act on the main label branch.
Neither method updates the label row object in your script. To see a task’s new stage, list the label row again with project.list_label_rows_v2().

# Import dependencies
from encord.user_client import EncordUserClient

#Authenticate using the path to your private key
user_client = EncordUserClient.create_with_ssh_private_key(
    ssh_private_key_path="<private_key_path>",
    # For US platform users use "https://api.us.encord.com"
    domain="https://api.encord.com",
    )

# Open the project you want to work on by specifying the project hash
project = user_client.get_project(
    "<project-hash>"
)

# Create a bundle and automatically execute everything attached to the bundle
with project.create_bundle() as bundle:
    # Return all data units in the task back to the first Workflow stage
    for label_row in project.list_label_rows_v2():
        label_row.workflow_reopen(bundle=bundle)

# Import dependencies
from encord.user_client import EncordUserClient

#Authenticate using the path to your private key
user_client = EncordUserClient.create_with_ssh_private_key(
    ssh_private_key_path="<private_key_path>",
    # For US platform users use "https://api.us.encord.com"
    domain="https://api.encord.com",
    )

# Open the project you want to work on by specifying the project hash
project = user_client.get_project(
    "<project-hash>"
)

# Get the label row for the specific data unit you want to reopen
label_row = project.list_label_rows_v2(data_hashes=["<data-unit-id>"])[0]

# Return the task to the first Workflow stage
label_row.workflow_reopen()

Move Tasks Freely

Moving tasks skips all intermediate Workflow stages, placing them immediately into the selected stage.
All labels (object and classification) are automatically approved when tasks move to a COMPLETE stage.
Webhooks do not fire when tasks teleport through a workflow.

This script shows how to move a data unit directly between workflow stages, bypassing the standard workflow progression. The script shows how to access various Workflow Stage types; you should only use the ones relevant to your project.

Moving Tasks

from encord import EncordUserClient
from encord.workflow import AnnotationStage, ReviewStage,  AgentStage, FinalStage, ConsensusAnnotationStage

PROJECT_ID = '<project_ID>'
DATA_HASH = '<data_unit_hash>'
user_client = EncordUserClient.create_with_ssh_private_key(
    ssh_private_key_path='<private_key_path>',
    # For US platform users use "https://api.us.encord.com"
    domain="https://api.encord.com",
)

project = user_client.get_project(PROJECT_ID)

# Get your required stages
consensus_annotation_stage = project.workflow.get_stage(name='Consensus 1', type_=ConsensusAnnotationStage)
annotation_stage = project.workflow.get_stage(name='Annotate 1', type_=AnnotationStage)
review_stage = project.workflow.get_stage(name='Review 1', type_=ReviewStage)
agent_stage = project.workflow.get_stage(name='Agent 1', type_=AgentStage)
complete_stage = project.workflow.get_stage(name='Complete', type_=FinalStage)
archive_stage = project.workflow.get_stage(name='Archive', type_=FinalStage)

# Insert source and destination stages here
source_stage = <insert_source_stage>
destination_stage = <insert_destination_stage>

# Move a task from the source stage to the destination stage
with project.create_bundle() as bundle:
    for task in source_stage.get_tasks(data_hash=DATA_HASH):
        print(f"Moving task from '{source_stage.title}' to '{destination_stage.title}'")
        task.move(destination_stage_uuid=destination_stage.uuid, bundle=bundle)

Task Priority

You can read and assign task priorities using the SDK. Priorities in the SDK are represented by numbers ranging from 0 to 1, where 1 corresponds to 100 in the Encord platform, the highest priority value.

For more information, see our platform documentation for the Queue tab, and priorities.

Setting priorities requires the Admin or Team Manager role on the Project. A bundle sends priorities in requests of up to 1,000 tasks, so you can set priorities for any number of tasks in one bundle.

# Import dependencies
from encord.user_client import EncordUserClient

#Authenticate using the path to your private key
user_client = EncordUserClient.create_with_ssh_private_key(
    ssh_private_key_path="<private_key_path>",
    # For US platform users use "https://api.us.encord.com"
    domain="https://api.encord.com",
    )

# Open the project you want to work on by specifying the project hash
project = user_client.get_project(
    "<project-hash>"
)

# Reading the priority of the first label row
label_row = project.list_label_rows_v2()[0]
label_row.initialise_labels()

print(f"Current priority: {label_row.priority}")

# Setting the priority of the first label row to 0.2
label_row.set_priority(0.2)
label_row = project.list_label_rows_v2()[0]
print(f"New priority: {label_row.priority}")

# Import dependencies
from encord.user_client import EncordUserClient

#Authenticate using the path to your private key
user_client = EncordUserClient.create_with_ssh_private_key(
    ssh_private_key_path="<private_key_path>",
    # For US platform users use "https://api.us.encord.com"
    domain="https://api.encord.com",
    )

# Open the project you want to work on by specifying the project hash
project = user_client.get_project(
    "<project-hash>"
)

# Create a bundle
with project.create_bundle() as bundle:

# Set the priority of all tasks in the project to 0.5 
  for label_row in project.list_label_rows_v2():
    label_row.set_priority(0.5, bundle=bundle)

End-to-End Example moving tasks through Projects

This example uses a basic non-Consensus Project Workflow moving a number of tasks from ANNOTATE to COMPLETE. !End-to-end Example Workflow !End-to-end Example tasks

Project ID: 01bd7084-b04a-4cd1-87f3-73e8e78925c4
Data units: apples_01.jpg, apples_02.jpg, apples_03.jpg, apples_04.jpg, apples_05.jpg

STEP 1: View all tasks in the Project

A few tasks have been annotated, but they have not been submitted yet. The following code lists all tasks in the ANNOTATE 1 stage.

# Import dependencies
from encord import EncordUserClient, Project

from encord.workflow import(
  AnnotationStage, 
  ReviewStage,
  ConsensusAnnotationStage, 
  ConsensusReviewStage,
  FinalStage
)


SSH_PATH = "/Users/chris-encord/sdk-ssh-private-key.txt"
PROJECT_HASH = "01bd7084-b04a-4cd1-87f3-73e8e78925c4"

user_client: EncordUserClient = EncordUserClient.create_with_ssh_private_key(
    ssh_private_key_path=SSH_PATH,
    # For US platform users use "https://api.us.encord.com"
    domain="https://api.encord.com",
)

# Get project
project: Project = user_client.get_project(PROJECT_HASH)

stage = project.workflow.get_stage(name="Annotate 1", type_=AnnotationStage)
# type_ is an optional parameter here, any other type narrowing technique can be used here
# for example it can be an assert like follows
assert isinstance(stage, AnnotationStage)

for task in stage.get_tasks():
    
    print(f"Task: {task}")

Task: uuid=UUID('387dbbe6-4abb-47f8-9efe-199d1d4652ce') created_at=datetime.datetime(2024, 6, 28, 17, 19, 10, 374492) updated_at=datetime.datetime(2024, 6, 30, 22, 51, 51, 139841) data_hash=UUID('34571f50-90d1-4681-a58b-1fd08e9f6665') data_title='apples_05.jpg' label_branch_name='main' assignee=None
Task: uuid=UUID('80e09443-f649-4624-9cff-222c04115d16') created_at=datetime.datetime(2024, 6, 28, 17, 19, 10, 374492) updated_at=datetime.datetime(2024, 6, 30, 22, 53, 58, 858513) data_hash=UUID('22f8dc9a-deff-49a2-a2d2-3863cf20d4f4') data_title='apples_03.jpg' label_branch_name='main' assignee='annotator003@encord.com'
Task: uuid=UUID('d5fa67aa-9490-4248-ad0d-287b5dc23189') created_at=datetime.datetime(2024, 6, 28, 17, 19, 10, 374492) updated_at=datetime.datetime(2024, 6, 30, 22, 52, 15, 119658) data_hash=UUID('809344fd-8935-4dde-81eb-36efb4e6d558') data_title='apples_01.jpg' label_branch_name='main' assignee='annotator001@encord.com'
Task: uuid=UUID('e53132b4-fd13-4673-bf65-41fe8e690125') created_at=datetime.datetime(2024, 6, 28, 17, 19, 10, 374492) updated_at=datetime.datetime(2024, 6, 30, 22, 55, 9, 87875) data_hash=UUID('4dc183ba-4233-4cb1-a940-cb6b2bc38f35') data_title='apples_04.jpg' label_branch_name='main' assignee='annotator-reviewer001@encord.com'
Task: uuid=UUID('e793424c-1d2f-4f32-aa18-920ada4b2a7a') created_at=datetime.datetime(2024, 6, 28, 17, 19, 10, 374492) updated_at=datetime.datetime(2024, 6, 30, 22, 52, 48, 962443) data_hash=UUID('36b0afd3-6ff8-4201-bf65-35a37f6fe5b5') data_title='apples_02.jpg' label_branch_name='main' assignee='annotator002@encord.com'

From the returned results we can see that four of the data units have been assigned (apples_01 to apples_04) while one task has not (apples_05). We do not know the extent that each of the assigned data units has been annotated, but we want the tasks to move through the Workflow.

STEP 2: Submit all tasks for review

The following code submits ALL tasks to the REVIEW 1 stage. Before the tasks can be submitted they are first assigned to a user in the Project. The user could be any user capable of annotating in the Project (because this is the ANNOTATE stage). For this example, the Admin of the Project (the user with SDK access) is assigned.

Submit all tasks for review

# Import dependencies
from encord import EncordUserClient, Project

from encord.workflow import(
  AnnotationStage, 
  ReviewStage,
  ConsensusAnnotationStage, 
  ConsensusReviewStage,
  FinalStage
)

SSH_PATH = "/Users/chris-encord/sdk-ssh-private-key.txt"
PROJECT_HASH = "01bd7084-b04a-4cd1-87f3-73e8e78925c4"

user_client: EncordUserClient = EncordUserClient.create_with_ssh_private_key(
    ssh_private_key_path=SSH_PATH,
    # For US platform users use "https://api.us.encord.com"
    domain="https://api.encord.com",
)

# Get project
project: Project = user_client.get_project(PROJECT_HASH)

stage = project.workflow.get_stage(name="Annotate 1", type_=AnnotationStage)
# type_ is an optional parameter. Any type of narrowing technique can be used here.
# For example:
assert isinstance(stage, AnnotationStage)

BUNDLE_SIZE = 100

with project.create_bundle(bundle_size=BUNDLE_SIZE) as bundle:
    for task in stage.get_tasks():

        task.assign("chris@acme.com")

        task.submit(bundle=bundle)
        
        print(f"Task: {task}")

STEP 3: Verify that the tasks are in REVIEW 1

Each time we perform an action, we should verify the action is successful. After submitting the tasks, we can verify that the tasks are now in REVIEW 1.

# Import dependencies
from encord import EncordUserClient, Project

from encord.workflow import(
  AnnotationStage, 
  ReviewStage,
  ConsensusAnnotationStage, 
  ConsensusReviewStage,
  FinalStage
)


SSH_PATH = "/Users/chris-encord/sdk-ssh-private-key.txt"
PROJECT_HASH = "01bd7084-b04a-4cd1-87f3-73e8e78925c4"

user_client: EncordUserClient = EncordUserClient.create_with_ssh_private_key(
    ssh_private_key_path=SSH_PATH,
    # For US platform users use "https://api.us.encord.com"
    domain="https://api.encord.com",
)

# Get project
project: Project = user_client.get_project(PROJECT_HASH)

stage = project.workflow.get_stage(name="Review 1", type_=ReviewStage)
# type_ is an optional parameter here, any other type narrowing technique can be used here
# for example it can be an assert like follows
assert isinstance(stage, ReviewStage)

for task in stage.get_tasks():
    
    print(f"Task: {task}")

Task: uuid=UUID('643dc436-3c90-478b-a7f6-45cd70d0fc19') created_at=datetime.datetime(2024, 6, 30, 22, 46, 57, 229922) updated_at=datetime.datetime(2024, 6, 30, 23, 15, 28, 115530) assignee=None data_hash=UUID('809344fd-8935-4dde-81eb-36efb4e6d558') data_title='apples_01.jpg'
Task: uuid=UUID('02ace181-87dd-4628-8b17-7f9f16015c43') created_at=datetime.datetime(2024, 6, 30, 23, 15, 26, 636022) updated_at=datetime.datetime(2024, 6, 30, 23, 15, 26, 709843) assignee=None data_hash=UUID('34571f50-90d1-4681-a58b-1fd08e9f6665') data_title='apples_05.jpg'
Task: uuid=UUID('6aae1e0b-8297-4efa-9274-e1355633010d') created_at=datetime.datetime(2024, 6, 30, 23, 15, 27, 488760) updated_at=datetime.datetime(2024, 6, 30, 23, 15, 27, 565142) assignee=None data_hash=UUID('22f8dc9a-deff-49a2-a2d2-3863cf20d4f4') data_title='apples_03.jpg'
Task: uuid=UUID('e9623d55-b166-4a7e-bf8e-4e3b635263db') created_at=datetime.datetime(2024, 6, 30, 23, 15, 29, 32894) updated_at=datetime.datetime(2024, 6, 30, 23, 15, 29, 149193) assignee=None data_hash=UUID('4dc183ba-4233-4cb1-a940-cb6b2bc38f35') data_title='apples_04.jpg'
Task: uuid=UUID('4ff5801d-ad1a-4304-8270-55be654d681a') created_at=datetime.datetime(2024, 6, 30, 23, 15, 30, 422) updated_at=datetime.datetime(2024, 6, 30, 23, 15, 30, 86008) assignee=None data_hash=UUID('36b0afd3-6ff8-4201-bf65-35a37f6fe5b5') data_title='apples_02.jpg'

All of the tasks are now in stage REVIEW 1. None of the tasks are assigned to any users.

STEP 4: Reject a task

We know that apples-05.jpg does not have any labels, because it was not assigned to any users before we submitted the task. If we REJECT the task, according to the Workflow in the Project, the task returns to the ANNOTATE 1 stage.The following code filters the task in the REVIEW 1 stage based on the task’s data hash, and then rejects the task.

# Import dependencies
from encord import EncordUserClient, Project
from encord.workflow import AnnotationStage, ReviewStage, ConsensusAnnotationStage, ConsensusReviewStage, FinalStage
from uuid import UUID  # Import the UUID function

# Constants
SSH_PATH = "/Users/chris-encord/sdk-ssh-private-key.txt"
PROJECT_HASH = "01bd7084-b04a-4cd1-87f3-73e8e78925c4"

# Create user client
user_client: EncordUserClient = EncordUserClient.create_with_ssh_private_key(
    ssh_private_key_path=SSH_PATH,
    # For US platform users use "https://api.us.encord.com"
    domain="https://api.encord.com",
)

# Get project
project: Project = user_client.get_project(PROJECT_HASH)

# Get stage
stage = project.workflow.get_stage(name="Review 1", type_=ReviewStage)
# Uncomment the following line for different stage

# Check instance type
assert isinstance(stage, ReviewStage)

data_hashes = ["34571f50-90d1-4681-a58b-1fd08e9f6665"]

BUNDLE_SIZE = 100

with project.create_bundle(bundle_size=BUNDLE_SIZE) as bundle:
    for task in stage.get_tasks(data_hash=data_hashes):

        task.assign("chris@acme.com")

        task.reject(bundle=bundle)

        print(f"Task: {task}")

Task: uuid=UUID('02ace181-87dd-4628-8b17-7f9f16015c43') created_at=datetime.datetime(2024, 6, 30, 23, 15, 26, 636022) updated_at=datetime.datetime(2024, 6, 30, 23, 15, 26, 709843) assignee=None data_hash=UUID('34571f50-90d1-4681-a58b-1fd08e9f6665') data_title='apples_05.jpg'

From what was returned, we can see that only data unit apples_05.jpg was acted upon.To verify that the task is now in ANNOTATE 1, run the following:

# Import dependencies
from encord import EncordUserClient, Project

from encord.workflow import(
  AnnotationStage, 
  ReviewStage,
  ConsensusAnnotationStage, 
  ConsensusReviewStage,
  FinalStage
)


SSH_PATH = "/Users/chris-encord/sdk-ssh-private-key.txt"
PROJECT_HASH = "01bd7084-b04a-4cd1-87f3-73e8e78925c4"

user_client: EncordUserClient = EncordUserClient.create_with_ssh_private_key(
    ssh_private_key_path=SSH_PATH,
    # For US platform users use "https://api.us.encord.com"
    domain="https://api.encord.com",
)

# Get project
project: Project = user_client.get_project(PROJECT_HASH)

stage = project.workflow.get_stage(name="Annotate 1", type_=AnnotationStage)
# type_ is an optional parameter here, any other type narrowing technique can be used here
# for example it can be an assert like follows
assert isinstance(stage, AnnotationStage)

for task in stage.get_tasks():
    
    print(f"Task: {task}")

Task: uuid=UUID('387dbbe6-4abb-47f8-9efe-199d1d4652ce') created_at=datetime.datetime(2024, 6, 28, 17, 19, 10, 374492) updated_at=datetime.datetime(2024, 6, 30, 23, 48, 3, 805733) data_hash=UUID('34571f50-90d1-4681-a58b-1fd08e9f6665') data_title='apples_05.jpg' label_branch_name='main' assignee='chris@acme.com'

STEP 5: Approve tasks

There are still four tasks in REVIEW 1. The following code approves the tasks. This moves the tasks to the COMPLETE stage. You could approve all the tasks at once the REVIEW 1 stage, but instead we will filter by data hash and then approve.

# Import dependencies
from encord import EncordUserClient, Project
from encord.workflow import AnnotationStage, ReviewStage, ConsensusAnnotationStage, ConsensusReviewStage, FinalStage
from uuid import UUID  # Import the UUID function

# Constants
SSH_PATH = "/Users/chris-encord/sdk-ssh-private-key.txt"
PROJECT_HASH = "01bd7084-b04a-4cd1-87f3-73e8e78925c4"

# Create user client
user_client: EncordUserClient = EncordUserClient.create_with_ssh_private_key(
    ssh_private_key_path=SSH_PATH,
    # For US platform users use "https://api.us.encord.com"
    domain="https://api.encord.com",
)

# Get project
project: Project = user_client.get_project(PROJECT_HASH)

# Get stage
stage = project.workflow.get_stage(name="Review 1", type_=ReviewStage)
# Uncomment the following line for different stage

# Check instance type
assert isinstance(stage, ReviewStage)

data_hashes = [
    "36b0afd3-6ff8-4201-bf65-35a37f6fe5b5",
    "4dc183ba-4233-4cb1-a940-cb6b2bc38f35",
    "22f8dc9a-deff-49a2-a2d2-3863cf20d4f4",
    "809344fd-8935-4dde-81eb-36efb4e6d558"
    ]



BUNDLE_SIZE = 100

with project.create_bundle(bundle_size=BUNDLE_SIZE) as bundle:
    for task in stage.get_tasks(data_hash=data_hashes):

        task.assign("chris@acme.com")

        task.approve(bundle=bundle)

        print(f"Task: {task}")

Task: uuid=UUID('643dc436-3c90-478b-a7f6-45cd70d0fc19') created_at=datetime.datetime(2024, 6, 30, 22, 46, 57, 229922) updated_at=datetime.datetime(2024, 6, 30, 23, 15, 28, 115530) assignee=None data_hash=UUID('809344fd-8935-4dde-81eb-36efb4e6d558') data_title='apples_01.jpg'
Task: uuid=UUID('6aae1e0b-8297-4efa-9274-e1355633010d') created_at=datetime.datetime(2024, 6, 30, 23, 15, 27, 488760) updated_at=datetime.datetime(2024, 6, 30, 23, 15, 27, 565142) assignee=None data_hash=UUID('22f8dc9a-deff-49a2-a2d2-3863cf20d4f4') data_title='apples_03.jpg'
Task: uuid=UUID('e9623d55-b166-4a7e-bf8e-4e3b635263db') created_at=datetime.datetime(2024, 6, 30, 23, 15, 29, 32894) updated_at=datetime.datetime(2024, 6, 30, 23, 15, 29, 149193) assignee=None data_hash=UUID('4dc183ba-4233-4cb1-a940-cb6b2bc38f35') data_title='apples_04.jpg'
Task: uuid=UUID('4ff5801d-ad1a-4304-8270-55be654d681a') created_at=datetime.datetime(2024, 6, 30, 23, 15, 30, 422) updated_at=datetime.datetime(2024, 6, 30, 23, 15, 30, 86008) assignee=None data_hash=UUID('36b0afd3-6ff8-4201-bf65-35a37f6fe5b5') data_title='apples_02.jpg'

STEP 6: Verify tasks are COMPLETE

Use the following code to verify that the tasks apples01 to apples_04 are in the COMPLETE stage.

# Import dependencies
from encord import EncordUserClient, Project

from encord.workflow import(
  AnnotationStage, 
  ReviewStage,
  ConsensusAnnotationStage, 
  ConsensusReviewStage,
  FinalStage
)


SSH_PATH = "/Users/chris-encord/sdk-ssh-private-key.txt"
PROJECT_HASH = "09765dc5-08b0-4b32-b37a-e1b932b8b684"

user_client: EncordUserClient = EncordUserClient.create_with_ssh_private_key(
    ssh_private_key_path=SSH_PATH,
    # For US platform users use "https://api.us.encord.com"
    domain="https://api.encord.com",
)

# Get project
project: Project = user_client.get_project(PROJECT_HASH)

stage = project.workflow.get_stage(name="Complete", type_=FinalStage)
# type_ is an optional parameter here, any other type narrowing technique can be used here
# for example it can be an assert like follows
assert isinstance(stage, FinalStage)

for task in stage.get_tasks():
    
    print(f"Task: {task}")

Task: uuid=UUID('e27acb09-0300-4bc9-8143-d0b837fd9207') created_at=datetime.datetime(2024, 7, 1, 0, 18, 22, 448867) updated_at=datetime.datetime(2024, 7, 1, 0, 18, 22, 495702) data_hash=UUID('22f8dc9a-deff-49a2-a2d2-3863cf20d4f4') data_title='apples_03.jpg'
Task: uuid=UUID('23351c59-32a6-42b9-8e06-fe20c41513bc') created_at=datetime.datetime(2024, 7, 1, 0, 18, 22, 67363) updated_at=datetime.datetime(2024, 7, 1, 0, 18, 22, 110112) data_hash=UUID('4dc183ba-4233-4cb1-a940-cb6b2bc38f35') data_title='apples_04.jpg'
Task: uuid=UUID('9e8fce04-af0e-41ca-926a-56bf354a2cba') created_at=datetime.datetime(2024, 7, 1, 0, 18, 21, 168387) updated_at=datetime.datetime(2024, 7, 1, 0, 18, 21, 223495) data_hash=UUID('36b0afd3-6ff8-4201-bf65-35a37f6fe5b5') data_title='apples_02.jpg'
Task: uuid=UUID('41d98c6d-2f4c-4ad1-8f91-576b74e56600') created_at=datetime.datetime(2024, 7, 1, 0, 18, 20, 516722) updated_at=datetime.datetime(2024, 7, 1, 0, 18, 20, 561088) data_hash=UUID('809344fd-8935-4dde-81eb-36efb4e6d558') data_title='apples_01.jpg'

And from what the code returns, the tasks are now in the COMPLETE stage.

Get Started

General

Data

Ontologies

Projects

Labels

DICOM

Workflows and Stages

Workflow Basics

View all stages in a Workflow

Non-Consensus Projects

Display Tasks in a Stage

Assign a User to Tasks in a Stage

Submit Tasks as Another User in a Stage

Retain Current User and Submit Task

Release Tasks in a Stage

Reopen Label Reviews in a Review Task

Get all Tasks

Filter tasks by data hash

Filter tasks by status

Move all tasks in a stage to the next stage

Move Agent tasks along a pathway

Use model predictions as an annotator

Consensus Projects

Get all Subtasks

Select your labels from a Consensus Project

Work with Labels on Tasks

Move Tasks

To Next Stage

To Complete

Reopen

Move Tasks Freely

Task Priority

End-to-End Example moving tasks through Projects