argo-workflows

import argo_workflows from argo_workflows.api import workflow_service_api import os # Configure API client configuration = argo_workflows.Configuration( host=os.environ.get("ARGO_WORKFLOWS_HOST", "http://localhost:2746"), verify_ssl=os.environ.get("ARGO_WORKFLOWS_VERIFY_SSL", "false").lower() == "true", ) # Bearer Token Authentication (common for Argo Workflows) token = os.environ.get("ARGO_WORKFLOWS_TOKEN") if token: configuration.api_key["Authorization"] = token configuration.api_key_prefix["Authorization"] = "Bearer" else: print("Warning: ARGO_WORKFLOWS_TOKEN not set. API calls might fail without authentication.") # Enter a context with an instance of the API client with argo_workflows.ApiClient(configuration) as api_client: # Create an instance of the API class api_instance = workflow_service_api.WorkflowServiceApi(api_client) namespace = os.environ.get("ARGO_WORKFLOWS_NAMESPACE", "argo") # Example namespace try: # Example: List workflows in a given namespace api_response = api_instance.list_workflows(namespace=namespace) print(f"Successfully listed {len(api_response.items)} workflows in namespace '{namespace}'.") if api_response.items: print(f"First workflow: {api_response.items[0].metadata.name}") except argo_workflows.ApiException as e: print(f"Exception when calling WorkflowServiceApi->list_workflows: {e}\n")