Data management in SDK

Overview

Manage geospatial data in your storage with PySTAC.

PySTAC authentication

After you’ve authenticated with the SDK, create a PySTAC client connection.

UP42_client = up42.stac_client()

The code samples on this page require an initialized UP42_client object.

Class: Collection

A PySTAC class that represents a STAC collection. See the PySTAC documentation for more information.

A STAC collection is a delivery you receive after a completed order or a finished processing job.

Attributes

Attribute	Description
`id`	str The STAC collection ID.
`title`	Optional[str] The title of the STAC collection.
`assets`	dict[str, Asset] A dictionary where keys are asset names and values are `Asset` objects linked to this STAC collection. A collection might optionally contain one asset: “Original Delivery”. If this asset isn’t present at the collection level, it’s stored in the individual STAC items instead.
`stac_extensions`	list[str] Enabled STAC extensions.

Properties

up42

Provides access to all UP42-specific metadata embedded within the collection.

1
import json
2

3
# Select a STAC collection
4
stac_collection_id = "a0d443a2-41e8-4995-8b54-a5cc4c448227"
5

6
# Fetch STAC collection info
7
collection = UP42_client.get_collection(stac_collection_id)
8
up42_properties = collection.up42
9

10
# Define output
11
readable_json = json.dumps(up42_properties._reference, indent=2)
12
print(readable_json)

Methods

get_collections

Retrieves all STAC collections. Returns Iterator[Collection]. Use itertools.islice to offset and limit the results. See the PySTAC documentation for more information.

1
from itertools import islice
2

3
# Search for STAC collections
4
collections = UP42_client.get_collections()
5

6
# Define output
7
for collection in islice(collections, 0, 5): # Print first 5 results
8
    print(f"- STAC collection ID:   {collection.id}")
9
    print(f"  Filename:             {collection.title}")
10
    print(f"  User title:           {collection.up42.title}")
11
    print(f"  Source:               {collection.up42.source}")
12
    for asset_key, asset_obj in collection.assets.items(): # Print the only asset if it exists
13
        print(f"  Asset key:            {asset_key}")
14
        print(f"  Asset title:          {asset_obj.title}")
15
        print(f"  Media type:           {asset_obj.media_type}")
16
        print(f"  Roles:                {', '.join(asset_obj.roles)}")
17
        print(f"  Link:                 {asset_obj.href}")
18
    print()

get_collection

Retrieves a specific STAC collection associated with an item. Returns Collection. See the PySTAC documentation for more information.

1
# Select a STAC item
2
stac_item_id = "a0d443a2-41e8-4995-8b54-a5cc4c448227"
3

4
# Fetch the parent STAC collection
5
collection = UP42_client.get_item(stac_item_id).get_collection()
6

7
# Define output
8
print(f"STAC collection ID: {collection.id}")
9
print(f"Filename:           {collection.title}")
10
print(f"User title:         {collection.up42.title}")
11
print(f"Source:             {collection.up42.source}")
12
for asset_key, asset_obj in collection.assets.items(): # Print the only asset if it exists
13
    print(f"Asset key:          {asset_key}")
14
    print(f"Asset title:        {asset_obj.title}")
15
    print(f"Media type:         {asset_obj.media_type}")
16
    print(f"Roles:              {', '.join(asset_obj.roles)}")
17
    print(f"Link:               {asset_obj.href}")

Class: Item

A PySTAC class that represents a STAC item. See the PySTAC documentation for more information.

A STAC item is an individual scene that represents inseparable data and metadata. A STAC item has a unique spatiotemporal extent, that is, has one AOI and acquisition time in its characteristics.

Attributes

Attribute	Description
`id`	str The STAC item ID.
`geometry`	Optional[Union[geojson.Polygon, geojson.MultiPolygon]] The STAC item geometry in the GeoJSON format.
`bbox`	Optional[BoundingBox] The bounding box that encloses the STAC item geometry.
`datetime`	Optional[str] The date and time when the sensor acquired the data.
`properties`	dict Additional STAC item metadata.
`assets`	dict[str, Asset] A dictionary where keys are asset names and values are `Asset` objects linked to this STAC item.
`stac_extensions`	list[str] Enabled STAC extensions.

Properties

up42

Provides access to all UP42-specific metadata embedded within the item.

1
import json
2

3
# Select a STAC item
4
stac_item_id = "a0d443a2-41e8-4995-8b54-a5cc4c448227"
5

6
# Fetch STAC item info
7
item = UP42_client.get_item(stac_item_id)
8
up42_properties = item.up42
9

10
# Define output
11
readable_json = json.dumps(up42_properties._reference, indent=2)
12
print(readable_json)

Methods

search

Retrieves STAC items matching search criteria. Returns ItemSearch. Use itertools.islice to offset and limit the results. See the PySTAC documentation for more information.

Parameter	Description
`filter`	Optional[SearchFilter] The CQL2 search filter.
`ids`	Optional[List[str]] The IDs of STAC items you want to include in search results.
`bbox`	Optional[BoundingBox] A search geometry in the GeoJSON format. Returns images that intersect with the defined rectangle and may not fully cover it. Use only if `intersects` isn’t specified.
`intersects`	Optional[geojson.Polygon] A polygon in the GeoJSON format.
`datetime`	Optional[str] The date and time when the sensor acquired the data.

1
from itertools import islice
2

3
# Define the search filter
4
filter = {
5
    "op": "=",
6
    "args": [
7
        {"property": "order_id"}, # This example searches by order ID
8
        "68567134-27ad-7bd7-4b65-d61adb11fc78", # The selected order ID
9
    ],
10
}
11

12
# Search for STAC items
13
items = UP42_client.search(filter=filter)
14

15
# Define output
15 collapsed lines
16
for item in islice(items.items(), 0, 5): # Print first 5 results
17
    print(f"- STAC item ID: {item.id}")
18
    print(f"  User title:   {item.up42.title}")
19
    print(f"  User tags:    {item.up42.tags}")
20
    print(f"  Source:       {item.up42.source}")
21
    print(f"  Job ID:       {item.up42.job_id or 'N/A'}") # For jobs
22
    print(f"  Order ID:     {item.up42.order_id or 'N/A'}") # For orders
23
    print(f"  Properties:   {item.properties}")
24
    for asset_key, asset_obj in islice(item.assets.items(), 0, 3): # Print first 3 assets in each item
25
        print(f"  - Asset key:    {asset_key}")
26
        print(f"    Asset title:  {asset_obj.title}")
27
        print(f"    Media type:   {asset_obj.media_type}")
28
        print(f"    Roles:        {', '.join(asset_obj.roles)}")
29
        print(f"    Link:         {asset_obj.href}")
30
    print()

get_items

Retrieves all STAC items. Returns Iterator[Item]. Use itertools.islice to offset and limit the results. See the PySTAC documentation for more information.

1
from itertools import islice
2

3
# Search for STAC items
4
items = UP42_client.get_items()
5

6
# Define output
15 collapsed lines
7
for item in islice(items, 0, 5): # Print first 5 results
8
    print(f"- STAC item ID: {item.id}")
9
    print(f"  User title:   {item.up42.title}")
10
    print(f"  User tags:    {item.up42.tags}")
11
    print(f"  Source:       {item.up42.source}")
12
    print(f"  Job ID:       {item.up42.job_id or 'N/A'}") # For jobs
13
    print(f"  Order ID:     {item.up42.order_id or 'N/A'}") # For orders
14
    print(f"  Properties:   {item.properties}")
15
    for asset_key, asset_obj in islice(item.assets.items(), 0, 3): # Print first 3 assets in each item
16
        print(f"  - Asset key:    {asset_key}")
17
        print(f"    Asset title:  {asset_obj.title}")
18
        print(f"    Media type:   {asset_obj.media_type}")
19
        print(f"    Roles:        {', '.join(asset_obj.roles)}")
20
        print(f"    Link:         {asset_obj.href}")
21
    print()

get_item

Retrieves a specific STAC item by its ID. Returns Item. See the PySTAC documentation for more information.

Parameter	Description
`id`	str The STAC item ID.

1
from itertools import islice
2

3
# Select a STAC item
4
stac_item_id = "a0d443a2-41e8-4995-8b54-a5cc4c448227"
5

6
# Fetch STAC item info
7
item = UP42_client.get_item(stac_item_id)
8

9
# Define output
13 collapsed lines
10
print(f"STAC item ID:   {item.id}")
11
print(f"User title:     {item.up42.title}")
12
print(f"User tags:      {item.up42.tags}")
13
print(f"Source:         {item.up42.source}")
14
print(f"Job ID:         {item.up42.job_id or 'N/A'}") # For jobs
15
print(f"Order ID:       {item.up42.order_id or 'N/A'}") # For orders
16
print(f"Properties:     {item.properties}")
17
for asset_key, asset_obj in islice(item.assets.items(), 0, 3): # Print first 3 assets
18
    print(f"  - Asset key:    {asset_key}")
19
    print(f"    Asset title:  {asset_obj.title}")
20
    print(f"    Media type:   {asset_obj.media_type}")
21
    print(f"    Roles:        {', '.join(asset_obj.roles)}")
22
    print(f"    Link:         {asset_obj.href}")

update

Changes the title and tags of a STAC item.

1
# Select a STAC item
2
stac_item_id = "a0d443a2-41e8-4995-8b54-a5cc4c448227"
3

4
# Fetch STAC item info
5
item = UP42_client.get_item(stac_item_id)
6

7
# Update the title in our database
8
item.up42.title = "Pleiades Berlin"
9
item.update()
10

11
# Define output to check the changes
12
print(f"STAC item ID:   {item.id}")
13
print(f"User title:     {item.up42.title}")

1
# Select a STAC item
2
stac_item_id = "a0d443a2-41e8-4995-8b54-a5cc4c448227"
3

4
# Fetch STAC item info
5
item = UP42_client.get_item(stac_item_id)
6

7
# Define new tags to add
8
new_tags = ["ortho-done"]
9

10
# Update tags in our database
11
item.up42.tags = item.up42.tags + new_tags # This adds new tags to existing ones
12
# item.up42.tags = ["only-ortho-done"] # Uncomment to overwrite existing tags
13
item.update()
14

15
# Define output to check the changes
16
print(f"STAC item ID:   {item.id}")
17
print(f"User tags:      {item.up42.tags}")

Class: Asset

A PySTAC class that represents a STAC asset. See the PySTAC documentation for more information.

A STAC asset is a geospatial feature of a STAC item. For example, a band in an optical image, a thumbnail, or a metadata file.

Attributes

Attribute	Description
`href`	str The URL of a downloadable or a streamable file related to the STAC asset.
`title`	Optional[str] The STAC asset title.
`description`	Optional[str] The STAC asset description.
`media_type`	Optional[str] The STAC asset data type.
`roles`	Optional[list[str]] The purpose of the STAC asset.

Properties

file.url

Retrieves the signed URL for the asset file. Returns str. Uses the file helper object.

1
# Select a STAC item
4 collapsed lines
2
stac_item_id = "a0d443a2-41e8-4995-8b54-a5cc4c448227"
3

4
# Fetch STAC item info
5
item = UP42_client.get_item(stac_item_id)
6

7
# Find the first asset with the 'multispectral' role
8
multispectral_asset = next(
9
  (asset for asset in item.assets.values() if asset.roles and "multispectral" in asset.roles),
10
  None,
11
)
12

13
# Define output
14
if multispectral_asset:
15
    print(f"The signed URL: {multispectral_asset.file.url}")
16
else:
17
    print("No asset found.")

Methods

file.download

Downloads assets from storage. Uses the file helper object.

Parameter Description

Parameter	Description
`output_directory`	Union[str, pathlib.Path] The file output directory.

output_directory

Union[str, pathlib.Path]

The file output directory.

Original deliveries

1
import pathlib
2

3
# Select a STAC item
5 collapsed lines
4
stac_item_id = "a0d443a2-41e8-4995-8b54-a5cc4c448227"
5

6
# Define a directory to store the downloaded ZIP
7
output_dir = pathlib.Path.home() / "Desktop" / "original_zipped_deliveries"
8
output_dir.mkdir(parents=True, exist_ok=True)
9

10
# Fetch the parent STAC collection
11
collection = UP42_client.get_item(stac_item_id).get_collection()
12

13
# Find the asset with the 'original' role
14
original_delivery = next(
15
    (asset for asset in collection.assets.values() if asset.roles and "original" in asset.roles),
16
    None,
17
)
18

19
# Download the file
20
if original_delivery:
21
    downloaded_file_path = original_delivery.file.download(output_directory=output_dir)
22
    print(f"\nThe original delivery downloaded to '{downloaded_file_path}'.")
23
else:
24
    print(f"The original delivery wasn't found.")

1
import pathlib
2

3
# Define the search filter
14 collapsed lines
4
filter = {
5
    "op": "=",
6
    "args": [
7
        {"property": "order_id"}, # This example searches by order ID
8
        "68567134-27ad-7bd7-4b65-d61adb11fc78", # The selected order ID
9
    ],
10
}
11

12
# Search for STAC items
13
items = UP42_client.search(filter=filter)
14

15
# Define a directory to store the downloaded files
16
output_dir = pathlib.Path.home() / "Desktop" / "original_zipped_deliveries"
17
output_dir.mkdir(parents=True, exist_ok=True)
18

19
# Fetch the parent STAC collections
20
for item in items.items():
21
    collection = item.get_collection()
22
    if not collection:
23
        continue
24

25
    # Find the assets with the 'original' role
26
    original_delivery = next(
27
        (asset for asset in collection.assets.values() if asset.roles and "original" in asset.roles),
28
        None,
29
    )
30

31
    # Download the file
32
    if original_delivery:
33
        downloaded_file_path = original_delivery.file.download(output_directory=output_dir)
34
        print(f"\nThe original delivery was downloaded to '{downloaded_file_path}'.")
35
        break
36
else:
37
    print(f"The original delivery wasn't found.")

1
import pathlib
2

3
# Define the search criteria
8 collapsed lines
4
orders = up42.Order.all(
5
    status=["FULFILLED"],
6
    tags=["project-7"],
7
)
8

9
# Define a directory to store the downloaded files
10
output_dir = pathlib.Path.home() / "Desktop" / "original_zipped_deliveries"
11
output_dir.mkdir(parents=True, exist_ok=True)
12

13
# Process each order
21 collapsed lines
14
for order in orders:
15
    print(f"- Order ID:         {order.id}")
16
    print(f"  Display name:     {order.display_name}")
17
    print(f"  Status:           {order.status}")
18
    print(f"  Order tags:       {', '.join(order.tags) if order.tags else 'No tags'}")
19

20
    # The filter searches for all matching orders
21
    filter = {
22
        "op": "=",
23
        "args": [
24
            {"property": "order_id"},
25
            order.id
26
        ],
27
    }
28
    items = UP42_client.search(filter=filter)
29

30
    found = False
31
    for item in items.items():
32
        collection = item.get_collection()
33
        if not collection:
34
            continue
35

36
        # Find the asset with the 'original' role
37
        original_delivery = next(
38
            (asset for asset in collection.assets.values()
39
             if asset.roles and "original" in asset.roles),
40
            None,
41
        )
42

43
        # Download the file
44
        if original_delivery:
45
            downloaded_file_path = original_delivery.file.download(output_directory=output_dir)
46
            print(f"The original delivery for order {order.id} downloaded to '{downloaded_file_path}'.\n")
47
            found = True
48
            break  # Only one 'original' asset per order
49
    if not found:
50
        print(f"The original delivery wasn't found for order {order.id}.")
51

52
print("All downloads complete.")

Geospatial assets

See all available roles.

1
import pathlib
2

3
# Select a STAC item and the asset role
6 collapsed lines
4
stac_item_id = "a0d443a2-41e8-4995-8b54-a5cc4c448227"
5
role = "data"
6

7
# Define a directory to store the downloaded files
8
output_dir = pathlib.Path.home() / "Desktop" / "data_files"
9
output_dir.mkdir(parents=True, exist_ok=True)
10

11
# Fetch STAC item info
12
item = UP42_client.get_item(stac_item_id)
13

14
# Find all assets with the selected role and download them
15
download_count = 0
16
for asset in item.assets.values():
17
    if asset.roles and role in asset.roles:
18
        asset.file.download(output_directory=output_dir)
19
        download_count += 1
20
        print()
21

22
# Check the count
23
if download_count > 0:
24
    print(f"\n{download_count} file(s) downloaded to '{output_dir}'.")
25
else:
26
    print("\nNo assets of this role were found.")

1
import pathlib
2

3
# Select a STAC item
6 collapsed lines
4
stac_item_id = "a0d443a2-41e8-4995-8b54-a5cc4c448227"
5
file_type = "image/tiff"
6

7
# Define a directory to store the downloaded files
8
output_dir = pathlib.Path.home() / "Desktop" / "tiff_files"
9
output_dir.mkdir(parents=True, exist_ok=True)
10

11
# Fetch STAC item info
12
item = UP42_client.get_item(stac_item_id)
13

14
# Find all assets of the selected format and download them
15
download_count = 0
16
for asset in item.assets.values():
17
    if asset.media_type and file_type in asset.media_type:
18
        asset.file.download(output_directory=output_dir)
19
        download_count += 1
20
        print()
21

22
# Check the count
23
if download_count > 0:
24
    print(f"\n{download_count} file(s) downloaded to '{output_dir}'.")
25
else:
26
    print("\nNo assets of this file format were found.")

1
import pathlib
2

3
# Select an order
14 collapsed lines
4
order_id = "68567134-27ad-7bd7-4b65-d61adb11fc78"
5

6
# Define a directory to store the downloaded files
7
output_dir = pathlib.Path.home() / "Desktop" / "order_assets"
8
output_dir.mkdir(parents=True, exist_ok=True)
9

10
# The search filter
11
filter = {
12
    "op": "=",
13
    "args": [
14
        {"property": "order_id"},
15
        order_id,
16
    ],
17
}
18

19
# Search for all STAC items from the order
20
items = UP42_client.search(filter=filter)
21

22
# Find and download all assets from all items from the order
23
download_count = 0
24
for item in items.items():
25
    print(f"\nProcessing assets for item {item.id}")
26
    for asset in item.assets.values():
27
        print(f"Downloading asset '{asset.title or asset.href}'")
28
        asset.file.download(output_directory=output_dir)
29
        download_count += 1
30
        print()
31

32
# Check the count
33
if download_count > 0:
34
    print(f"{download_count} file(s) downloaded to '{output_dir}'.")
35
else:
36
    print(f"No items or assets were found for order ID: '{order_id}'.")

1
import pathlib
2

3
# Select a job
14 collapsed lines
4
job_id = "55434287-31bc-3ad7-1a63-d61aac11ac55"
5

6
# Define a directory to store the downloaded files
7
output_dir = pathlib.Path.home() / "Desktop" / "job_assets"
8
output_dir.mkdir(parents=True, exist_ok=True)
9

10
# The search filter
11
filter = {
12
    "op": "=",
13
    "args": [
14
        {"property": "job_id"},
15
        job_id,
16
    ],
17
}
18

19
# Search for all STAC items from the job
20
items = UP42_client.search(filter=filter)
21

22
# Find and download all assets from all items from the job
23
download_count = 0
24
for item in items.items():
25
    print(f"\nProcessing assets for item {item.id}")
26
    for asset in item.assets.values():
27
        print(f"Downloading asset '{asset.title or asset.href}'")
28
        asset.file.download(output_directory=output_dir)
29
        download_count += 1
30
        print()
31

32
# Check the count
33
if download_count > 0:
34
    print(f"{download_count} file(s) downloaded to '{output_dir}'.")
35
else:
36
    print(f"No items or assets were found for job ID: '{job_id}'.")

Class: Up42Extension

A class that defines a namespace to access data from UP42 STAC extensions.

Attributes

Attribute	Description
`title`	str Refers to the `up42-user:title` property.
`tags`	str Refers to the `up42-user:tags` property.
`product_id`	str Refers to the `up42-product:product_id` property.
`collection_name`	str Refers to the `up42-product:collection_name` property.
`modality`	str Refers to the `up42-product:modality` property.
`order_id`	str Refers to the `up42-order:id` property.
`asset_id`	str Refers to the `up42-system:asset_id` property.
`account_id`	str Refers to the `up42-system:account_id` property.
`workspace_id`	str Refers to the `up42-system:workspace_id` property.
`job_id`	str Refers to the `up42-system:job_id` property.
`source`	str Refers to the `up42-system:source` property.
`metadata_version`	str Refers to the `up42-system:metadata_version` property.

Class: BulkDeletion

A class that deletes entire STAC collections by gathering all of their STAC item IDs into a deletion batch.

All items from a collection must be added for the operation to be successful.

Attributes

Attribute Description

Attribute	Description
`item_ids`	str, … STAC item IDs to add to the deletion batch. Each ID must be passed separately.

item_ids

str, …

STAC item IDs to add to the deletion batch. Each ID must be passed separately.

1
# Select STAC items to delete
2
item_ids_to_delete = [
3
    "14371401-97a3-4b23-9b55-c3ac10d47750",
4
    "ea36dee9-fed6-457e-8400-2c20ebd30f44",
5
]
6

7
# Initialize the bulk deletion handler and add items
8
deleter = UP42_client.BulkDeletion(*item_ids_to_delete)
9

10
# Submit the deletion request
11
deleter.delete()

Methods

delete

Submits the batch of staged STAC items to be deleted.

An item can only be deleted if all items of the same STAC collection are added for deletion.

1
# Select STAC items to delete
2
item_ids_to_delete = [
3
    "14371401-97a3-4b23-9b55-c3ac10d47750",
4
    "ea36dee9-fed6-457e-8400-2c20ebd30f44",
5
]
6

7
# Initialize the bulk deletion handler and add items
8
deleter = UP42_client.BulkDeletion(*item_ids_to_delete)
9

10
# Submit the deletion request
11
deleter.delete()

Data management in SDK

Use data management functionalities of the SDK using PySTAC.

Overview

PySTAC authentication

Class: Collection

Attributes

Properties

up42

Methods

get_collections

get_collection

Class: Item

Attributes

Properties

up42

Methods

search

get_items

get_item

update

Class: Asset

Attributes

Properties

file.url

Methods

file.download

Original deliveries

Geospatial assets

Class: Up42Extension

Attributes

Class: BulkDeletion

Attributes

Methods

delete

Learn more

API docs | Data management