Data management in SDK

Overview

Manage geospatial data in your storage with PySTAC.

View repository

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 = next(UP42_client.get_items(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 = next(UP42_client.get_items(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 STAC items. Returns Iterator[Item]. Use itertools.islice to offset and limit the results. See the PySTAC documentation for more information.

Parameter Description

Parameter	Description
`*ids`	Optional[str, …] STAC item IDs.

*ids

Optional[str, …]

STAC item IDs.

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()

update

This action can’t be done with a Viewer role.

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 = next(UP42_client.get_items(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 = next(UP42_client.get_items(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 = next(UP42_client.get_items(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 = next(UP42_client.get_items(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 = next(UP42_client.get_items(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 = next(UP42_client.get_items(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`	List[str] STAC item IDs to add to the deletion batch. Each ID must be passed separately.

item_ids

List[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.BulkDeletion(*item_ids_to_delete)
9

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

Methods

delete

This action can’t be done with a Viewer role.

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.BulkDeletion(*item_ids_to_delete)
9

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

Learn more

API docs | Data management

Console docs | Data management

Recently selected

Recommended articles

Search results

No results found

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

update

Class: Asset

Attributes

Properties

file.url

Methods

file.download

Original deliveries

Geospatial assets

Class: Up42Extension

Attributes

Class: BulkDeletion

Attributes

Methods

delete

Learn more