dtc-query-client 1.0.4

DTC Query API

DTC-IS Query Client

This package provides a Python client for the DTC Ice Sheets Query API.

DTC Ice Sheets is a Digital Twin Component designed to provide an integrated, interactive, and holistic representation of the Greenland and Antarctic ice sheet systems and their two-way interaction with the Earth system: ocean, atmosphere and solid Earth. By combining Earth Observation data, numerical models, and data science and machine learning methods into a modular, executable framework, it enables users to analyse past and present conditions, simulate future scenarios, and explore the sensitivity, response, and impact of ice-sheets under different conditions. Developed under ESA's Digital Twin Earth programme, the platform delivers these capabilities as trusted, API-first services.

Queries require authentication via an API token, which can be obtained at https://query.dtc-ice-sheets.org/auth/get-token

Refer to the API documentation for a complete list of available endpoints and their parameters.

For more information about the project, please visit https://dtc-ice-sheets.org/.

Installation

Install this package from PyPI:

pip install dtc-query-client

Getting Started

Start by importing the dtc_query_client and creating a Configuration object, passing in your authentication token:

import dtc_query_client

configuration = dtc_query_client.Configuration(
    host="https://query.dtc-ice-sheets.org",
    access_token="[your private token here]"
)

API requests are made inside an async with block. For example, this snippet retrieves a list of available datasets:

async with dtc_query_client.ApiClient(configuration) as client:
    datasets = await dtc_query_client.GenericApi(client).dataset_overviews()

print(datasets)

Some endpoints start a long-running job and return a job ID. This job ID can be used to track how the job is progressing, and to obtain results once the job completes. It is recommended to use the wait_for_job() or iterate_job() helper functions to handle these endpoints gracefully.

The example snippet below uses the simpler wait_for_job(), which waits for the job to complete and returns the completed job state:

async with dtc_query_client.ApiClient(configuration) as client:
    # Start a job and get the job ID
    response = await dtc_query_client.StateAndFateApi(client).run_summary_zarr_to_json(
        ice_shelf_id="thwaites",
    )
    job_id = response.job_id
    print(f"Started job with ID {job_id}")
    
    # Wait for job to complete
    try:
        job_state = await dtc_query_client.wait_for_job(client, job_id)
    except RuntimeError as e:
        # Failed jobs will raise a RuntimeError
        print("Job has failed!")
        raise e
    
    # Job has completed! Print results
    print(f"Job completed successfully with status {job_state.status}.")
    output_url = job_state.outputs["timeseries-zarr-to-json-module"]["output_json"]
    print(f"Results URL: {output_url}")

Alternatively, the example snippet below uses the more advanced iterate_job(), which does the same thing but additionally provides live updates on job progress while it is running:

async with dtc_query_client.ApiClient(configuration) as client:
    # Start a job and get the job ID
    response = await dtc_query_client.StateAndFateApi(client).run_summary_zarr_to_json(
        ice_shelf_id="thwaites",
    )
    job_id = response.job_id
    print(f"Started job with ID {job_id}")
    
    # Wait for job to complete, printing information on the job state as it progresses
    try:
        async for job_state in dtc_query_client.iterate_job(client, job_id):
            print(
                f"Job has status {job_state.status}.\n"
                f"It has completed {job_state.progress_done} steps out of {job_state.progress_total}."
            )
    except RuntimeError as e:
        # Failed jobs will raise a RuntimeError
        print("Job has failed!")
        raise e
    
    # Job has completed! Print results
    print(f"Job completed successfully with status {job_state.status}.")
    output_url = job_state.outputs["timeseries-zarr-to-json-module"]["output_json"]
    print(f"Results URL: {output_url}")