Quickstart

pip install --upgrade scaleapi

import scaleapi

API_KEY = "live_...."
client = scaleapi.ScaleClient(API_KEY)

# Downloads all completed tasks from a project and saves them to a jsonl file
import json

PROJECT_NAME = "My Test Project"

def get_tasks_by_project_name(project_name: str):
    tasks_generator = client.v2_get_tasks(
        project_name=project_name,
        status="completed",
        limit=100,
    )
    tasks = list(tasks_generator)
    return tasks

if __name__ == "__main__":
    tasks = get_tasks_by_project_name(PROJECT_NAME)

    with open(f"{PROJECT_NAME}.jsonl", "w+") as f:
        for task in tasks:
            json.dump(task.model_dump(mode='json'), f)
            f.write("\n")

from scaleapi.exceptions import (
    ScaleException,
    ScaleResourceNotFound,
)

def get_or_create_batch(client: scaleapi.ScaleClient, project_name, batch_name):
    try:
        batch = client.get_batch(batch_name)
        return batch
    except ScaleResourceNotFound:
        batch = client.create_batch(project_name, batch_name, "")
        return batch
    except ScaleException as err:
        raise Exception(f"❌ ERROR Batch Creation: {err.message}") from err

batch = get_or_create_batch(client, project.name, "test_api_batch_20240708")
print(f"Batch: {batch.name}")

# Those variables are specific to each project and need to be aligned with Scale before sending them
template_variables = {
    "id": "0",
    "prompt": "Hello world!",
    "base_model_response": "..."
    "test_modeL_response": "..."
    "languageCode": "en_US",
}

# Task Payload
payload = {
    "project": project.name,
    "batch": batch.name,
    "template_variables": template_variables,
    "unique_id":"unique_id_12345"
}

def create_task_request(client: scaleapi.ScaleClient, payload):
    try:
        task = client.create_task(TaskType.Chat, **payload)
        return task.id
    except ScaleDuplicateResource as dup_err:  # Existing Task
        print(f"Task already exists: {dup_err.message}")
    except ScaleException as err:
        print(f"❌ ERROR Task Creation: {err.message}")
        return None

task_id = create_task_request(client, payload)