Python client for Taskcluster

99.0.3 · active · verified Sun Apr 12

The `taskcluster` library is a comprehensive Python client for interacting with the Taskcluster API. It provides both synchronous and asynchronous interfaces for all Taskcluster API methods, allowing users to define, schedule, and manage tasks within the Taskcluster continuous integration and continuous delivery (CI/CD) system. Currently at version 99.0.3, the library maintains a rapid release cadence, with its version numbers co-versioned with the main Taskcluster platform.

Warnings

Install

Imports

Quickstart

This quickstart demonstrates how to initialize a synchronous Taskcluster client, typically configured via environment variables. It attempts to fetch the status of the Taskcluster Queue service as a basic connectivity test. It also shows how to import the asynchronous client module. For authentication, `TASKCLUSTER_ROOT_URL`, `TASKCLUSTER_CLIENT_ID`, and `TASKCLUSTER_ACCESS_TOKEN` environment variables are commonly used.

import os
import taskcluster

# Configure client using environment variables (recommended)
options = taskcluster.optionsFromEnvironment()

# Alternatively, explicit configuration
# options = {
#     'rootUrl': 'https://taskcluster.example.com',
#     'credentials': {
#         'clientId': os.environ.get('TASKCLUSTER_CLIENT_ID', 'your-client-id'),
#         'accessToken': os.environ.get('TASKCLUSTER_ACCESS_TOKEN', 'your-access-token')
#     }
# }

# Create a synchronous client for the Queue service
queue = taskcluster.Queue(options)

# Example: List tasks (requires appropriate scopes and a running Taskcluster instance)
try:
    # This is a simplified example; actual API calls vary.
    # For a real listing, you'd likely use `listTasks` with filters.
    # We'll just try to get a service status as a quick test.
    status = queue.status()
    print(f"Taskcluster Queue Service Status: {status['state']}")
except Exception as e:
    print(f"Could not connect to Taskcluster or fetch status: {e}")
    print("Ensure TASKCLUSTER_ROOT_URL and credentials are set if required.")

# For asynchronous operations, use the aio client:
# import asyncio
# async def main():
#     aio_queue = taskcluster.aio.Queue(options)
#     status = await aio_queue.status()
#     print(f"Async Queue Service Status: {status['state']}")
# asyncio.run(main())

view raw JSON →