Google Cloud Orchestration Airflow API Client Library

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)