Example Usage

This page provides reusable snippets to authenticate, search the catalog, order, task, search tasking requests, consume imagery and metadata and more.

from capella_console_client import CapellaConsoleClient

authenticate

with API key

client = CapellaConsoleClient(api_key="<api-key>", verbose=True)

# don't want to validate the api-key (saves an API call)
client = CapellaConsoleClient(api_key="<api-key>", no_token_check=True)

# prompts for API key
client = CapellaConsoleClient()

# reads API key from CAPELLA_API_KEY env
os.environ["CAPELLA_API_KEY"] = "<api-key>"   # NOTE: provide valid API Key
client = CapellaConsoleClient()

with access token

# already have a valid access token (JWT)? no problem
client = CapellaConsoleClient(token="<token>", verbose=True)

# don't want to validate the token (saves an API call)
client = CapellaConsoleClient(token="<token>", no_token_check=True)

catalog

group search results

results = client.search(
    instrument_mode="spotlight",
    product_type="GEO",
    sortby="-datetime"
)

by_stac_id = results.groupby(field="id")

by_collect_id = results.groupby(field="collect_id")

by_stac_collection = results.groupby(field="collection")

by_instrument_mode = results.groupby(field="instrument_mode")

by_instrument = res.groupby(field="instruments").keys()

search results

visualize search results

from pathlib import Path
import json

results = client.search(
    instrument_mode="spotlight",
    product_type="GEO",
    sortby="-datetime"
)
# store stac items in geojson FeatureCollection
feature_collection = results.to_feature_collection()

# write to disk
feature_collection_path = Path('CAPELLA_SP_GEOs.geojson')
feature_collection_path.write_text(json.dumps(feature_collection))

# open e.g. in QGIS

order

Issue the following snippets to submit a (purchasing) order by providing STAC items or STAC ids.

order items

# submit order with stac items
order_id = client.submit_order(items=capella_spotlight_olympic_NP_geo)

# alternatively order by STAC ids
first_two_ids = [item["id"] for item in capella_spotlight_olympic_NP_geo[:2]]
order_id = client.submit_order(stac_ids=first_two_ids)

# since orders expire you can alternatively check prior if an active order already exists
# instead of creating a new order - charges won't be applied twice anyways
order_id = client.submit_order(items=capella_spotlight_olympic_NP_geo,
                               check_active_orders=True)

download assets

Download assets of previously ordered products to local disk.

# download all products of an order to /tmp
product_paths = client.download_products(
    order_id=order_id,
    local_dir="/tmp",
)

# 🕒 don't like parallel downloads? 🕒 - set threaded = False in order to fetch the product assets serially
product_paths = client.download_products(
    order_id=order_id,
    local_dir="/tmp",
    threaded=False
)

# ⌛ like to watch progress bars? ⌛ - set show_progress = True in order to get feedback on download status (time remaining, transfer stats, ...)
product_paths = client.download_products(
    order_id=order_id,
    local_dir="/tmp",
    show_progress=True,
)

# the client is respectful of your local files and does not override them by default
# but can be instructed to do so
local_thumb_path = client.download_products(
    order_id=order_id,
    local_dir="/tmp",
    show_progress=True,
    override=True
)

Output

2021-06-21 20:28:16,734 - 🛰️  Capella 🐐 - INFO - downloading product CAPELLA_C03_SP_SLC_HH_20210621202423_20210621202425 to /tmp/CAPELLA_C03_SP_SLC_HH_20210621202423_20210621202425
CAPELLA_C03_SP_GEO_HH_20210603175705_20210603175729_thumb.png       ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ 100.0% • 211.3/211.3 KB   • 499.7 kB/s  • 0:00:00
CAPELLA_C03_SP_GEO_HH_20210619045726_20210619045747_thumb.png       ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ 100.0% • 307.1/307.1 KB   • 1.4 MB/s    • 0:00:00
CAPELLA_C03_SP_GEO_HH_20210619180117_20210619180140_thumb.png       ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ 100.0% • 271.6/271.6 KB   • 1.1 MB/s    • 0:00:00
CAPELLA_C03_SP_GEO_HH_20210627180259_20210627180321_extended.json   ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ 0.0%   • 20,426/-1 bytes  • 200.2 kB/s  • 0:00:00
CAPELLA_C03_SP_GEO_HH_20210603175705_20210603175729_extended.json   ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ 0.0%   • 21,536/-1 bytes  • 293.8 kB/s  • 0:00:00
CAPELLA_C03_SP_GEO_HH_20210619180117_20210619180140_extended.json   ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ 0.0%   • 20,650/-1 bytes  • 122.0 kB/s  • 0:00:00
CAPELLA_C03_SP_GEO_HH_20210627180259_20210627180321_thumb.png       ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ 100.0% • 316.7/316.7 KB   • 1.3 MB/s    • 0:00:00
CAPELLA_C03_SP_GEO_HH_20210603175705_20210603175729.tif             ━╸━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ 5.6%   • 13.2/237.4 MB    • 2.2 MB/s    • 0:01:42
CAPELLA_C03_SP_GEO_HH_20210619045726_20210619045747_extended.json   ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ 0.0%   • 22,002/-1 bytes  • 196.9 kB/s  • 0:00:00
CAPELLA_C03_SP_GEO_HH_20210627180259_20210627180321.tif             ━╺━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ 3.0%   • 11.0/360.9 MB    • 1.9 MB/s    • 0:03:04
CAPELLA_C03_SP_GEO_HH_20210619045726_20210619045747.tif             ╸━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ 2.7%   • 9.8/359.0 MB     • 1.8 MB/s    • 0:03:18

By default the respective product assets are saved into separate product directories, i.e.

/tmp/<stac_id_1>/<stac_id_1>.tif
/tmp/<stac_id_1>/<stac_id_1>_thumb.png
/tmp/<stac_id_1>/<stac_id_1>_extended.json
/tmp/<stac_id_2>/<stac_id_2>.tif
...

If you prefer a flat hierarchy set separate_dirs to False:

product_paths = client.download_products(
    order_id=order_id,
    separate_dirs=False,
)

single asset

single assets can be downloaded to gven paths

# download thumbnail
thumb_presigned_href = assets_presigned[0]["thumbnail"]["href"]
dest_path = "/tmp/thumb.png"
local_thumb_path = client.download_asset(thumb_presigned_href, local_path=dest_path)

# assets are saved into OS specific temp directory if `local_path` not provided
raster_presigned_href = assets_presigned[0]["HH"]["href"]
local_raster_path = client.download_asset(raster_presigned_href)


from pathlib import Path
assert local_thumb_path == Path(dest_path)

resume download

Automatic resume for Interrupted Downloads

Downloads automatically resume from the last byte if interrupted (network failure, timeout, user cancellation), which is enabled by default.

# Start downloading a large file
client.download_asset(
    presigned_url,
    local_path="/tmp/img.tif",
    show_progress=True
)
# ... download interrupted at 60% ...

# download resumes from 60% when running the same command again
client.download_asset(
    presigned_url,
    local_path="/tmp/img.tif",
    show_progress=True
)

Note

  • Downloads use HTTP Range headers to resume from the last already downloaded byte

  • If the server doesn’t support Range headers, the file is re-downloaded from the start

  • Complete files are skipped entirely (no unnecessary downloads)

  • Destination S3 paths have limited resume support (logged warning if detected)

  • Unknown file sizes skip resume for safety

  • Resume works for both single assets and batch product downloads:

# All products resume automatically if interrupted
product_paths = client.download_products(
    order_id=order_id,
    local_dir="/tmp",
    show_progress=True,
    enable_resume=True  # default
)

order filters

by product type

# download only GEO product
product_paths = client.download_products(
   order_id=order_id,
   product_types=["GEO"]
)

# download only SLC and GEO product
product_paths = client.download_products(
   order_id=order_id,
   product_types=["SLC", "GEO"]
)

by asset type

# download only thumbnails
product_paths = client.download_products(
   order_id=order_id,
   include=["thumbnail"]
)

# 'include' / 'exclude' can also be a string if only one provided
product_paths = client.download_products(
   order_id=order_id,
   include="thumbnail"
)

# download only raster (VV or HH)
product_paths = client.download_products(
   order_id=order_id,
   include="raster"
)

# download all assets except raster
product_paths = client.download_products(
   order_id=order_id,
   exclude="raster"
)

# explicit DENY overrides explicit ALLOW --> the following would only fetch thumbnails
product_paths = client.download_products(
   order_id=order_id,
   include=["raster", "thumbnail"]
   exclude="raster"
)

items of tasking request

Requirement: you have previously issued a tasking request that is in ‘completed’ state

tasking_request_id = "27a71826-7819-48cc-b8f2-0ad10bee0f97"  # NOTE: provide valid tasking_request_id

# download ALL products
product_paths = client.download_products(
    tasking_request_id=tasking_request_id,
)

# download only GEO product
product_paths = client.download_products(
    tasking_request_id=tasking_request_id,
    product_types=["GEO"]
)

items of collect

collect_id = "27a71826-7819-48cc-b8f2-0ad10bee0f97"  # NOTE: provide valid collect_id

# download ALL products
product_paths = client.download_products(
    collect_id=collect_id,
)

# download only GEC product
product_paths = client.download_products(
    collect_id=collect_id,
    product_types=["GEC"],
)

review order

If you would like to review the cost of an order before you submission, issue:

order_details = client.review_order(items=capella_spotlight_olympic_NP_geo)
print(order_details['orderDetails']['summary'])

presign assets

In order to directly load assets (imagery or other) into memory you need to request signed S3 URLs first.

items_presigned = client.get_presigned_items(order_id)

# alternatively presigned assets can also be filtered - e.g. give me the presigned assets of 2 specific STAC ids
first_two_ids = [item["id"] for item in capella_spotlight_olympic_NP_geo[:2]]
items_presigned = client.get_presigned_items(order_id,
                                               stac_ids=first_two_ids)

# sort presigned assets by list of stac ids
sorted_stac_ids = sorted([s['id'] for s in capella_spotlight_olympic_NP_geo])
items_presigned_sorted = client.get_presigned_items(order_id,
                                                    sort_by=sorted_stac_ids)

See read imagery or read metadata for more information.

list orders

Issue the following snippet to view the ordering history

# list all orders
all_orders = client.list_orders()

# list all active orders
all_active_orders = client.list_orders(is_active=True)

# list specific order(s) by order id
specific_order_id = all_orders[0]["orderId"]
specific_orders = client.list_orders(order_ids=[specific_order_id])

tasking requests

create

NOTE: geometry and name are the only required properties to create a tasking request.

# create point tasking request
client.create_tasking_request(
    geometry=geojson.Point([11.148216220469152, 49.59672249842626]),
    name="point tasking request #127"
)

# area tasking request
client.create_tasking_request(
    geometry=geojson.Polygon(
        [
            [
                [11.148216220469152, 49.59672249842626],
                [11.148216220469152, 49.55415435337187],
                [11.219621049225651, 49.55415435337187],
                [11.219621049225651, 49.59672249842626],
                [11.148216220469152, 49.59672249842626],
            ]
        ]
    ),
    name="area tasking request #127",
    collection_type="stripmap_100",
)

# tasking request customization
client.create_tasking_request(
    geometry=geojson.Polygon(
        [
            [
                [11.148216220469152, 49.59672249842626],
                [11.148216220469152, 49.55415435337187],
                [11.219621049225651, 49.55415435337187],
                [11.219621049225651, 49.59672249842626],
                [11.148216220469152, 49.59672249842626],
            ]
        ]
    ),
    name="highly customizable #127",
    description="too many knobs",
    collection_tier="urgent",
    collection_type="spotlight_ultra",
    local_time="day",
    off_nadir_min=5,
    off_nadir_max=50,
    orbitalPlanes=[45, 53],
    asc_dsc="ascending",
    look_direction="right",
    polarization="HH",
    archive_holdback="30 day",
    custom_attribute_1="correlation #1",
    custom_attribute_2="correlation #2",
    pre_approval=True,
    azimuth_angle_min=340,
    azimuth_angle_max=20,
    squint="enabled",
    max_squint_angle=25,
)


# same as above but leveraging enums defined in `enumerations.py` (client-side validation)
from capella_console_client.enumerations import (
    CollectionTier,
    CollectionType,
    LocalTimeOption,
    OrbitalPlane,
    OrbitState,
    Polarization,
    ObservationDirection,
    ArchiveHoldback,
    SquintMode,
)

client.create_tasking_request(
    geometry=geojson.Point([11.148216220469152, 49.59672249842626]),
    name="highly customizable #127",
    description="too many knobs",
    collection_tier=CollectionTier.urgent,
    collection_type=CollectionType.SPOTLIGHT_ULTRA,
    local_time=LocalTimeOption.day,
    off_nadir_min=5,
    off_nadir_max=50,
    orbital_planes=[OrbitalPlane.fortyfive, OrbitalPlane.fiftythree],
    asc_dsc=OrbitState.ascending,
    look_direction=ObservationDirection.right,
    polarization=Polarization.HH,
    archive_holdback=ArchiveHoldback.thirty_day,
    custom_attribute_1="correlation #1",
    custom_attribute_2="correlation #2",
    pre_approval=True,
    azimuth_angle_min=340,
    azimuth_angle_max=20,
    squint=SquintMode.ENABLED,
    max_squint_angle=30,
)

update

The following fields can be updated on an existing tasking request: name, description, custom_attribute_1, custom_attribute_2, product_types.

Pass one or more tasking request IDs — all receive the same update. Results are returned as a dict keyed by tasking request ID.

# update a single tasking request
result = client.update_tasking_requests(
    "aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa",
    name="updated name",
    description="updated description",
)
updated_tr = result["aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa"]

# update multiple tasking requests in parallel — all get the same fields applied
result = client.update_tasking_requests(
    "aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa",
    "bbbbbbbb-bbbb-bbbb-bbbb-bbbbbbbbbbbb",
    "cccccccc-cccc-cccc-cccc-cccccccccccc",
    custom_attribute_1="campaign-2026",
)

# on success each value is the updated tasking request dict
# on failure each value is {"success": False, "error": {...}}
for tr_id, tr_result in result.items():
    if tr_result.get("success") is False:
        print(f"{tr_id} failed: {tr_result['error']['code']}")
    else:
        print(f"{tr_id} updated: {tr_result['properties']['taskingrequestName']}")

cancel

Find more information here. For Cancellation fees please refer to Capella’s Tasking Cancellation Policy Overview.

# provide 1..N valid tasking request ids to be cancelled
cancel_result_by_id = sit_client.cancel_tasking_requests(
    "aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa",
    "bbbbbbbb-bbbb-bbbb-bbbb-bbbbbbbbbbbb",
    "cccccccc-cccc-cccc-cccc-cccccccccccc"
)

print(cancel_result_by_id)

repeat requests

create

NOTE: geometry and name are the only required properties to create a repeat request.

# create above tasking request as repeat series
from capella_console_client.enumerations import (
    RepeatCollectionTier,
    CollectionType,
    LocalTimeOption,
    OrbitalPlane,
    OrbitState,
    Polarization,
    ObservationDirection,
    ArchiveHoldback,
    SquintMode,
)

client.create_repeat_request(
    geometry=geojson.Point([11.148216220469152, 49.59672249842626]),
    name="highly customizable repeat request #127",
    description="too many knobs",
    collection_tier=RepeatCollectionTier.flexible,
    collection_type=CollectionType.SPOTLIGHT_ULTRA,
    local_time=LocalTimeOption.day,
    off_nadir_min=5,
    off_nadir_max=50,
    orbital_planes=[OrbitalPlane.fortyfive, OrbitalPlane.fiftythree],
    asc_dsc=OrbitState.ascending,
    look_direction=ObservationDirection.right,
    polarization=Polarization.HH,
    archive_holdback=ArchiveHoldback.thirty_day,
    custom_attribute_1="correlation #1",
    custom_attribute_2="correlation #2",
    azimuth_angle_min=340,
    azimuth_angle_max=20,
    squint=SquintMode.ENABLED,
    max_squint_angle=30,
)

repeat requests repeat cadence can be configured in multiple ways

# A) until cancelled, e.g. weekly starting now (defaults)
client.create_repeat_request(
    geometry=geojson.Point([11.148216220469152, 49.59672249842626]),
    name="daily repeat",
)

# B) start + end datetime + frequency , e.g. daily for 27 days
from datetime import datetime, timezone, timedelta
from capella_console_client.enumerations import RepeatCycle

repeat_start = datetime.now(tz=timezone.utc)
repeat_end = repeat_start + timedelta(days=27)

client.create_repeat_request(
    geometry=geojson.Point([11.148216220469152, 49.59672249842626]),
    name="daily repeat",
    description="daily repeat",
    repeat_start=repeat_start,
    repeat_end=repeat_end,
    repetition_interval=RepeatCycle.DAILY,
)

# C) number of collects + frequency, e.g. 5 collects weekly starting from now
client.create_repeat_request(
    geometry=geojson.Point([11.148216220469152, 49.59672249842626]),
    name="repeat five weeks",
    description="repeat five weeks",
    repeat_start=repeat_start,
    repetition_count=5,
    repetition_interval=RepeatCycle.WEEKLY,
)

update

The following fields can be updated on an existing repeat request: name, description, custom_attribute_1, custom_attribute_2, product_types.

Pass one or more repeat request IDs — all receive the same update. Results are returned as a dict keyed by repeat request ID.

# update a single repeat request
result = client.update_repeat_requests(
    "aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa",
    name="updated name",
    description="updated description",
)
updated_rr = result["aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa"]

# update multiple repeat requests in parallel — all get the same fields applied
result = client.update_repeat_requests(
    "aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa",
    "bbbbbbbb-bbbb-bbbb-bbbb-bbbbbbbbbbbb",
    "cccccccc-cccc-cccc-cccc-cccccccccccc",
    custom_attribute_1="campaign-2026",
)

# on success each value is the updated repeat request dict
# on failure each value is {"success": False, "error": {...}}
for rr_id, rr_result in result.items():
    if rr_result.get("success") is False:
        print(f"{rr_id} failed: {rr_result['error']['code']}")
    else:
        print(f"{rr_id} updated: {rr_result['properties']['repeatrequestName']}")

cancel

Find more information here. For Cancellation fees please refer to Capella’s Tasking Cancellation Policy Overview.

# provide 1..N valid repeat request ids to be cancelled
cancel_result_by_id = sit_client.cancel_repeat_requests(
    "aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa",
    "bbbbbbbb-bbbb-bbbb-bbbb-bbbbbbbbbbbb",
    "cccccccc-cccc-cccc-cccc-cccccccccccc"
)

print(cancel_result_by_id)

search

A multitude of query fields and query operators are supported.

repeat_request_id = "37a71826-7819-48cc-b8f2-0ad10bee0f97"  # provide valid repeat_request_id

# get repeat request by id
rr = client.search_repeat_request(repeat_request_id=repeat_request_id)

rr[0]

advanced repeat request search

# get ALL completed repeat requests of user
user_completed_rrs_result = client.search_repeat_request(status="completed")

# get all COMPLETED tasking requests of ORG (requires org manager/ admin role)
org_completed_trs_result = client.search_repeat_request(
    for_org=True,
    status="completed"
)

# get all completed tasking requests of org SUBMITTED AFTER 2022-12-01 (UTC)
org_completed_rrs_submitted_dec_22_result = client.search_repeat_request(
    for_org=True,
    status="completed",
    submission_time__gt="2022-12-1"
)

# subset of supported filters
active_sp_routine_rrs_result = client.search_repeat_request(
    status="active",
    repeat_start__lt="2025-10-01",
    repeat_start__gt="2025-06-01",
    collection_type=["spotlight", "spotlight_ultra"],
    collection_tier="routine"
)

# ⌛ like to watch progress bars? ⌛ - set show_progress = True to display an interactive progress bar during pagination
client.search_repeat_requests(
    status="active",
    for_org=True,
    show_progress=True
)

# progress bar works with both threaded and sequential modes
client.search_repeat_requests(
    status="active",
    for_org=True,
    threaded=False,
    show_progress=True
)

Output with progress bar

⠋ Fetching repeat requests... ━━━━━━━━━━━━━━━━━━ 2/3 67%

insar

create

import geojson
from capella_console_client.enumerations import InsarOrbit

# create insar repeat request with start/ end date, MIO orbit (2.95 days revisit)

client.create_repeat_request(
    geometry=geojson.Point([11.148216220469152, 49.59672249842626]),
    name="stripmap_100 insar repeat request",
    collection_tier="insar",
    collection_type="stripmap_100",
    product_types=["SLC", "GEO"],
    squint="disabled",
    repeat_start="2026-05-01",
    repeat_end="2026-12-01",
    insar_orbit=InsarOrbit.MIO_53_2P95,
)

# create spotlight insar repeat request with repetition count, SSO orbit (7 days revisit)

client.create_repeat_request(
    geometry=geojson.Point([11.148216220469152, 49.59672249842626]),
    name="spotlight insar repeat request",
    collection_tier="insar",
    collection_type="spotlight",
    product_types=["SLC", "GEO"],
    squint="disabled",
    repetition_count=10,
    insar_orbit=InsarOrbit.SSO_97_7P00,
)

Update, search and cancel operations are identical to non-insar repeat requests. See the repeat requests section above for examples.

consume

read imagery

Given a presigned asset href (see presigned items) load imagery into memory

NOTE: requires rasterio (not part of this package)

import rasterio

# raster profile
raster_presigned_href = assets_presigned[0]["HH"]["href"]
with rasterio.open(raster_presigned_href) as ds:
    print(ds.profile)

# read chunk of raster
with rasterio.open(raster_presigned_href) as ds:
    chunk = ds.read(1, window=rasterio.windows.Window(2000, 2000, 7000, 7000))
print(chunk.shape)

# read thumbnail
thumb_presigned_href = assets_presigned[0]["thumbnail"]["href"]
with rasterio.open(thumb_presigned_href) as ds:
    thumb = ds.read(1)
print(thumb.shape)

read metadata

import httpx

# read extended metadata .json
metadata_presigned_href = assets_presigned[0]["metadata"]["href"]
metadata = httpx.get(metadata_presigned_href).json()