AnnotationClientMixin

Provides annotation and task management operations for the Synapse backend.

Overview

The AnnotationClientMixin handles all operations related to tasks, annotations, projects, and task tagging. This mixin is automatically included in the BackendClient and provides methods for annotation workflows.

Project Operations

get_project(pk)

Retrieve detailed information about a specific project.

project = client.get_project(123)
print(f"Project: {project['name']}")
print(f"Description: {project['description']}")

Parameters:

• pk (int): Project ID

Returns:

• dict: Project details including configuration, metadata, and settings

Task Operations

get_task(pk, params)

Get detailed information about a specific task.

# Basic task details
task = client.get_task(456)

# Task with expanded data unit
task = client.get_task(456, params={'expand': 'data_unit'})

# Task with multiple expansions
task = client.get_task(456, params={
    'expand': ['data_unit', 'assignment', 'annotations']
})

Parameters:

• pk (int): Task ID
• params (dict): Query parameters for filtering and expansion

Common params:

• expand: List or string of related objects to include
• include_annotations: Whether to include annotation data

annotate_task_data(pk, data)

Submit annotation data for a task.

# Submit bounding box annotations
annotation_data = {
    'annotations': [
        {
            'type': 'bbox',
            'coordinates': [10, 10, 100, 100],
            'label': 'person',
            'confidence': 0.95
        },
        {
            'type': 'polygon',
            'points': [[0, 0], [50, 0], [50, 50], [0, 50]],
            'label': 'vehicle'
        }
    ],
    'metadata': {
        'annotator_id': 'user123',
        'timestamp': '2023-10-01T12:00:00Z'
    }
}

result = client.annotate_task_data(456, annotation_data)

Parameters:

• pk (int): Task ID
• data (dict): Annotation data structure

Returns:

• dict: Updated task with submitted annotations

list_tasks(params=None, url_conversion=None, list_all=False)

List tasks with filtering and pagination support.

# List tasks for a specific project
tasks = client.list_tasks(params={'project': 123})

# List tasks with status filter
tasks = client.list_tasks(params={
    'project': 123,
    'status': 'pending'
})

# Get all tasks (handles pagination automatically)
all_tasks = client.list_tasks(list_all=True)

# List tasks with custom URL conversion for files
tasks = client.list_tasks(
    params={'project': 123},
    url_conversion={'files': lambda url: f"https://cdn.example.com{url}"}
)

Parameters:

• params (dict, optional): Filtering parameters
• url_conversion (dict, optional): Custom URL conversion for file fields
• list_all (bool): If True, automatically handles pagination to get all results

Common filtering params:

• project: Filter by project ID
• status: Filter by task status (pending, in_progress, completed)
• assignee: Filter by assigned user ID
• created_after: Filter by creation date
• search: Text search in task content

Returns:

• tuple: (tasks_list, total_count) if list_all=False
• list: All tasks if list_all=True

create_tasks(data)

Create one or more new tasks.

# Create single task
new_task = client.create_tasks({
    'project': 123,
    'data_unit': 789,
    'priority': 'high',
    'metadata': {'batch': 'batch_001'}
})

# Create multiple tasks
new_tasks = client.create_tasks([
    {'project': 123, 'data_unit': 789},
    {'project': 123, 'data_unit': 790},
    {'project': 123, 'data_unit': 791}
])

Parameters:

• data (dict or list): Task data or list of task data

Task data structure:

• project (int, required): Project ID
• data_unit (int, required): Data unit ID
• priority (str, optional): Task priority (low, normal, high)
• assignee (int, optional): User ID to assign task to
• metadata (dict, optional): Additional task metadata

Returns:

• dict or list: Created task(s) with generated IDs

set_tags_tasks(data, params=None)

Set tags for multiple tasks in batch.

# Add tags to multiple tasks
client.set_tags_tasks({
    'ids': [456, 457, 458],
    'tags': [1, 2, 3],
    'action': 'add'
})

# Remove tags from tasks
client.set_tags_tasks({
    'ids': [456, 457],
    'tags': [1, 2],
    'action': 'remove'
})

Parameters:

• data (dict): Batch tagging data
• params (dict, optional): Additional parameters

Data structure:

• ids (list): List of task IDs to modify
• tags (list): List of tag IDs to apply or remove
• action (str): Operation type - 'add' or 'remove'

Task Tag Operations

get_task_tag(pk)

Get details about a specific task tag.

tag = client.get_task_tag(123)
print(f"Tag: {tag['name']} - {tag['description']}")

Parameters:

• pk (int): Tag ID

Returns:

• dict: Tag details including name, description, and meta

list_task_tags(params)

List available task tags with filtering.

# List all tags
tags = client.list_task_tags({})

# List tags for a specific project
project_tags = client.list_task_tags({
    'project': 123
})

# Search tags by name
search_tags = client.list_task_tags({
    'search': 'quality'
})

Parameters:

• params (dict): Filtering parameters

Common filtering params:

• project: Filter by project ID
• search: Text search in tag names
• color: Filter by tag color

Returns:

• tuple: (tags_list, total_count)

Example Workflows

Complete Annotation Workflow

from synapse_sdk.clients.backend import BackendClient

client = BackendClient(
    base_url="https://api.synapse.sh",
    access_token="your-access-token"
)

# 1. Get project details
project = client.get_project(123)
print(f"Working on project: {project['name']}")

# 2. List pending tasks
pending_tasks = client.list_tasks(params={
    'project': 123,
    'status': 'pending'
})

# 3. Process first task
if pending_tasks[0]:
    task = pending_tasks[0][0]  # First task from results
    task_id = task['id']

    # Get detailed task info
    detailed_task = client.get_task(task_id, params={'expand': 'data_unit'})

    # Submit annotations
    annotations = {
        'annotations': [
            {
                'type': 'bbox',
                'coordinates': [10, 10, 100, 100],
                'label': 'person'
            }
        ]
    }

    result = client.annotate_task_data(task_id, annotations)
    print(f"Annotations submitted for task {task_id}")

    # Add quality tag
    quality_tags = client.list_task_tags({'search': 'quality'})
    if quality_tags.get('results'):
        tag_id = quality_tags['results'][0]['id']
        client.set_tags_tasks({
            'ids': [task_id],
            'tags': [tag_id],
            'action': 'add'
        })

Batch Task Creation

# Create tasks for multiple data units
data_units = [789, 790, 791, 792, 793]

tasks_data = []
for data_unit_id in data_units:
    tasks_data.append({
        'project': 123,
        'data_unit': data_unit_id,
        'priority': 'normal',
        'metadata': {
            'batch': 'automated_batch_001',
            'source': 'data_import'
        }
    })

# Create all tasks in one request
created_tasks = client.create_tasks(tasks_data)
print(f"Created {len(created_tasks)} tasks")

# Get IDs of created tasks
task_ids = [task['id'] for task in created_tasks]

# Apply initial tags
initial_tags = client.list_task_tags({'search': 'new'})
if initial_tags.get('results'):
    tag_id = initial_tags['results'][0]['id']
    client.set_tags_tasks({
        'ids': task_ids,
        'tags': [tag_id],
        'action': 'add'
    })

Error Handling

from synapse_sdk.clients.exceptions import ClientError

try:
    task = client.get_task(999999)
except ClientError as e:
    if e.status_code == 404:
        print("Task not found")
    elif e.status_code == 403:
        print("Permission denied")
    else:
        print(f"API Error: {e}")

See Also

• BackendClient - Main backend client
• HITLClientMixin - Human-in-the-loop operations
• DataCollectionClientMixin - Data management