google-cloud-orchestration-airflow

import os from google.cloud.orchestration.airflow.service.v1 import EnvironmentsClient from google.cloud.orchestration.airflow.service.v1 import types def list_composer_environments(project_id: str, location: str): """Lists Cloud Composer environments in a given project and location.""" client = EnvironmentsClient() parent = f"projects/{project_id}/locations/{location}" try: request = types.ListEnvironmentsRequest(parent=parent) page_result = client.list_environments(request=request) print(f"Cloud Composer Environments in {location}:") found_any = False for environment in page_result: print(f" - {environment.name} (State: {environment.state.name})") found_any = True if not found_any: print(" No environments found.") except Exception as e: print(f"Error listing environments: {e}") if __name__ == "__main__": project_id = os.environ.get('GOOGLE_CLOUD_PROJECT', 'your-gcp-project-id') location = os.environ.get('GOOGLE_CLOUD_LOCATION', 'us-central1') # e.g., 'us-central1', 'europe-west1' if project_id == 'your-gcp-project-id': print("Please set the GOOGLE_CLOUD_PROJECT environment variable or replace 'your-gcp-project-id'.") elif location == 'us-central1': print("Please set the GOOGLE_CLOUD_LOCATION environment variable or choose your desired location.") else: list_composer_environments(project_id, location)