Tracking asynchronous Earth Engine exports¶

Demonstrates the earthlens.gee.jobs surface — the same job-tracking shape earthlens.ecmwf.jobs exposes for CDS, applied to GEE batch tasks.

Two tiny Drive exports are submitted with wait_for_export=False (so download() does NOT block), then:

1. list_recent_tasks enumerates them.
2. get_task_status fetches one by id.
3. cancel_task cancels both so they don't actually run.
4. wait_for_task_id polls one of them to its terminal CANCELLED state.

No Drive permission is needed — the tasks are cancelled before EE attempts to write.

Setup¶

Consolidate the imports and read GEE_SERVICE_ACCOUNT / GEE_SERVICE_KEY from the environment. earthlens.gee exposes the four job-tracking helpers used below.

In [ ]:

import os
import time
from pathlib import Path

from earthlens import EarthLens
from earthlens.gee import (
    cancel_task,
    get_task_status,
    list_recent_tasks,
    wait_for_task_id,
)

OUT_DIR = Path('out') / 'track-batch-exports'
OUT_DIR.mkdir(parents=True, exist_ok=True)
SERVICE_ACCOUNT = os.environ['GEE_SERVICE_ACCOUNT']
SERVICE_KEY = os.environ['GEE_SERVICE_KEY']
print(f'output directory: {OUT_DIR.resolve()}')

import os import time from pathlib import Path from earthlens import EarthLens from earthlens.gee import ( cancel_task, get_task_status, list_recent_tasks, wait_for_task_id, ) OUT_DIR = Path('out') / 'track-batch-exports' OUT_DIR.mkdir(parents=True, exist_ok=True) SERVICE_ACCOUNT = os.environ['GEE_SERVICE_ACCOUNT'] SERVICE_KEY = os.environ['GEE_SERVICE_KEY'] print(f'output directory: {OUT_DIR.resolve()}')

Submit two tiny Drive exports (non-blocking)¶

Build and authenticate the request¶

Build the GEE request first. wait_for_export=False is the key knob: download() returns a TaskInfo per export the moment the task is queued, instead of blocking on completion. Construction and authenticate() are kept on separate statements so each step is easy to read and re-run.

In [ ]:

gee = EarthLens(
    data_source="gee",
    start='2024-06-01',
    end='2024-06-02',
    dataset='UCSB-CHG/CHIRPS/DAILY',
    variables=['precipitation'],
    cadence='daily',
    aoi=[31.1, 29.9, 31.3, 30.1],
    path=str(OUT_DIR),
    scale=5566.0,
    export_via='drive',
    drive_folder='earthlens_track_demo',
    wait_for_export=False,  # <- the key knob
)
gee.authenticate(service_account=SERVICE_ACCOUNT, service_key=SERVICE_KEY)

gee = EarthLens( data_source="gee", start='2024-06-01', end='2024-06-02', dataset='UCSB-CHG/CHIRPS/DAILY', variables=['precipitation'], cadence='daily', aoi=[31.1, 29.9, 31.3, 30.1], path=str(OUT_DIR), scale=5566.0, export_via='drive', drive_folder='earthlens_track_demo', wait_for_export=False, # <- the key knob ) gee.authenticate(service_account=SERVICE_ACCOUNT, service_key=SERVICE_KEY)

Submit the exports¶

We pick two short date windows so EE creates two separate buckets. Calling download() queues both exports and returns their TaskInfo records immediately, each with its task id and READY state.

In [ ]:

submitted = gee.download(progress_bar=False)
print(f'submitted {len(submitted)} task(s):')
for info in submitted:
    print(f'  {info.id:26} {info.state:14} {info.description}')

submitted = gee.download(progress_bar=False) print(f'submitted {len(submitted)} task(s):') for info in submitted: print(f' {info.id:26} {info.state:14} {info.description}')

List recent tasks on the project¶

list_recent_tasks wraps ee.data.listOperations() with client-side filtering by state / age / task type / description prefix. Filtering by our description_prefix= matches what GEE._api emits as the task description: <asset-slug>_<bands>_<YYYYMMDD>.

In [ ]:

ours = list_recent_tasks(
    description_prefix='UCSB-CHG_CHIRPS_DAILY_precipitation',
    max_age_min=10,
)
print(f'{len(ours)} matching task(s):')
for t in ours:
    print(f'  {t.id:26} {t.state:14} {t.task_type:14} {t.description}')

ours = list_recent_tasks( description_prefix='UCSB-CHG_CHIRPS_DAILY_precipitation', max_age_min=10, ) print(f'{len(ours)} matching task(s):') for t in ours: print(f' {t.id:26} {t.state:14} {t.task_type:14} {t.description}')

Inspect one task in detail¶

get_task_status(id) issues a single ee.data.getOperation call and returns the same normalised TaskInfo.

In [ ]:

first = submitted[0]
detail = get_task_status(first.id)
print(detail.model_dump_json(indent=2))

first = submitted[0] detail = get_task_status(first.id) print(detail.model_dump_json(indent=2))

Cancel both tasks¶

cancel_task(id) wraps the modern ee.data.cancelOperation (replacing the deprecated cancelTask). Cancellation is asynchronous — EE acknowledges the request immediately and the state transitions to CANCEL_REQUESTED then CANCELLED shortly after.

In [ ]:

for info in submitted:
    cancel_task(info.id)
    print(f'cancel requested for {info.id}')

for info in submitted: cancel_task(info.id) print(f'cancel requested for {info.id}')

Wait for one task to reach a terminal state¶

wait_for_task_id(id) polls until the task reaches COMPLETED / FAILED / CANCELLED / CANCEL_REQUESTED. Since we cancelled the tasks above, the call below should return quickly with a RuntimeError reporting the cancelled state.

(The notebook catches that error so the cell still produces clean output — in a real script you'd let it propagate.)

In [ ]:

try:
    final = wait_for_task_id(submitted[0].id, poll_seconds=3, progress_bar=False)
    print(f'finished: {final.state}')
except RuntimeError as exc:
    print(f'expected (we cancelled it): {exc}')

try: final = wait_for_task_id(submitted[0].id, poll_seconds=3, progress_bar=False) print(f'finished: {final.state}') except RuntimeError as exc: print(f'expected (we cancelled it): {exc}')

What's next¶

• resolve_destination(info) returns the Drive / GCS / asset destination URIs of a completed task — for our cancelled ones, the destinations are empty.
• Catalog.list_recent_tasks / Catalog.get_task_status are the same calls bound on the catalog object, mirroring the ECMWF Catalog.list_recent_jobs ergonomic.
• The CLI python -m earthlens.gee.jobs {list,status,cancel,wait} exposes the same surface for shell scripts.