taskcluster

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())