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: 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 if found
20
if original_delivery:
21
    downloaded_file_path = original_delivery.file.download(output_directory=output_dir)
22
    print(f"\nThe original delivery downloaded to '{output_dir}'.")
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 files if found
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 '{output_dir}'.")
35
        break
36
else:
37
    print(f"The original delivery wasn't found.")

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.

Data management in SDK

Use data management functionalities of the SDK using PySTAC.

Overview

PySTAC authentication

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

Learn more

API docs | STAC