Python client for Taskcluster
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
- breaking The Docker image tags for `taskcluster/websocktunnel` now require a `v` prefix (e.g., `v99.0.0` instead of `99.0.0`).
- breaking Generic Worker now correctly evaluates absolute paths specified in `mounts` (properties `directory` and `file`) and `artifacts` (property `path`) within task payloads. Previously, these were treated as relative paths.
- breaking The Generic Worker configuration properties `deploymentId` and `checkForNewDeploymentEverySecs` are no longer supported and must be removed from configurations. Additionally, Generic Worker exit code 70 now signifies 'Worker Manager advised termination' instead of 'non-current deployment ID'.
Install
-
pip install taskcluster
Imports
- taskcluster
import taskcluster
- taskcluster.aio
import taskcluster.aio
- Queue
from taskcluster import Queue
Quickstart
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())