2. The Database: pgstac - eoAPI Workshop

pgstac is a PostgreSQL extension that enables STAC metadata management in a PostgreSQL database. eoAPI is useful to many organizations because the other components are configured to work seamlessly with STAC metadata that is stored in your pgstac database.

pypgstac is a Python package for interacting with a pgstac database. You will learn how to use pypgstac to perform the following operations on a pgstac database:

Generate STAC collection record
Add the record to the collections table with Loader.load_collections
Generate STAC item records
Add new records to the items table with Loader.load_items
Delete an item from the items table

For production deployments your STAC metadata generation and ingestion workflow will probably not take place in a notebook but the basic steps will be the same!

Fill in the input boxes below to get started with your own personal Sentinel-2 STAC collection!

You will be using the username value to create a unique STAC collection
Set a location that is special to you in the lat/lon field - this will determine the spatial extent of your STAC collection
You will need to enter the database credentials here in order to post data to the database in this notebook.

Workshop Setup¶

Run this cell to fetch database credentials. You’ll be prompted to enter the workshop access token provided by your instructor.

Note: API endpoints (STAC, Raster, Vector) are already configured in your environment.

from workshop_setup import setup, get_random_point

# This will prompt for your workshop token and fetch database credentials
config = setup()
default_lon, default_lat = get_random_point()

Now you can chose a unique username and a location from which you will generate your own personal Sentinel-2 STAC Collection. Feel free to use the default values, but if you are interested in looking at some satellite imagery from a particular place, this is your chance to pick one!

import os

import ipywidgets as widgets
from haikunator import Haikunator
from IPython.display import display

username_input = widgets.Text(
    value=Haikunator().haikunate(),
    placeholder="Enter your username",
    description="username:",
    disabled=False,
)

lat_input = widgets.BoundedFloatText(
    value=default_lat,
    min=-90,
    max=90,
    placeholder="enter the latitude of your hometown",
    description="latitude:",
    disabled=False,
)
lon_input = widgets.BoundedFloatText(
    value=default_lon,
    min=-180,
    max=180,
    placeholder="enter the longitude of your hometown",
    description="longitude:",
    disabled=False,
)

# Display the widgets
display(username_input)
display(lat_input)
display(lon_input)

2.1 Load a collection object¶

We’ll start by creating a collection, that will eventually contain items, within ~2 degrees of your provided location between January 1 2025 and April 18 2025.

import pystac_client
from pystac import Collection, Extent, SpatialExtent, TemporalExtent
from pystac.utils import str_to_datetime
from pypgstac.db import PgstacDB
from pypgstac.load import Loader, Methods
from shapely.geometry import Point

stac_api_endpoint = os.getenv("STAC_API_ENDPOINT")

collection_id = f"{username_input.value}-sentinel-2-c1-l2a"
bbox = Point(lon_input.value, lat_input.value).buffer(2).bounds

temporal_extent = [
    str_to_datetime("2025-01-01T00:00:00Z"),
    str_to_datetime("2025-04-18T00:00:00Z"),
]

my_collection = Collection(
    id=collection_id,
    description=f"{username_input.value}'s personal Sentinel-2 L2A collection",
    extent=Extent(
        spatial=SpatialExtent([[*bbox]]),
        temporal=TemporalExtent([temporal_extent]),
    ),
)
my_collection

To connect to the pgstac database and upload data pypgstac will use the PG* environment variables that are set in the docker network. With those set you can instantiate a PgstacDB and Loader pair.

db = PgstacDB()
loader = Loader(db)

The load_collections method accepts an iterable of STAC collection dictionaries, a file path to a collection.json file, or a .ndjson file with multiple collection records. You already have your pystac.Collection in memory in this session so you can just write it to a dictionary and upload it (in a list, i.e. iterable object).

The upsert method will add your collection to the collections table if it does not exist and update any fields with new values if the record does already exist.

loader.load_collections([my_collection.to_dict()], insert_mode=Methods.upsert)

Now if you run a query on the database you will find a record for your collection!

list(db.query(f"SELECT id from collections where id = '{my_collection.id}';"))

2.2 Upload items¶

You already learned how to create STAC items from scratch in the STAC metadata notebook so you get to take a shortcut for this one. Instead of creating STAC metadata from scratch you can just copy some from an existing STAC collection - this will be sufficient for our exercises during the workshop.

This code will find Sentinel-2 L2A items for the first few months of 2025 that intersect the bounding box of a 2 degree radius around the coordinate that you entered at the top of the notebook.

source_client = pystac_client.Client.open("https://earth-search.aws.element84.com/v1")

search = source_client.search(
    collections="sentinel-2-c1-l2a",
    bbox=bbox,
    datetime=temporal_extent,
    limit=100,  # pagination limit
)

items = search.item_collection()

print(len(items))

You are going to upload these items to the items table in the pgstac database but to do so you need to make sure the items’ collection ID matches an existing collection - set it to match the collection that you uploaded in the previous step.

# override the collection id to match your new collection
for item in items:
    item.set_collection(my_collection)

items[0]

Now you can use the load_items method to upload the list of STAC item dictionaries to the items table in the pgstac database. The insert_ignore method will upload any items that do not exist in the table and skip records that already exist.

loader.load_items([item.to_dict() for item in items], insert_mode=Methods.insert_ignore)

Hooray you uploaded some items! Now run a query to verify that the records landed in the items table.

n_items = db.query_one(
    f"SELECT COUNT(*) FROM items where collection = '{my_collection.id}';"
)
print(f"there are {n_items} items")

Here is your first glimpse of the power of STAC metadata. You just uploaded these items to the database and now you can browse the data in a beautiful interface without doing any more work!

Radiant Earth built a tool called STAC Browser that is a human-readable interface to a STAC API. Your collection is immediately availble in the STAC API so you can browse it in STAC Browser. Take a tour around the STAC Browser page for your collection.

from IPython.display import IFrame

# Use the stack's own STAC Browser when deployed (its catalog is already this
# STAC API); otherwise fall back to the public STAC Browser in external mode.
stac_browser_endpoint = os.getenv("STAC_BROWSER_ENDPOINT")

if stac_browser_endpoint:
    browser_url = f"{stac_browser_endpoint}/#/collections/{my_collection.id}"
else:
    browser_stac_url = os.getenv("STAC_API_ENDPOINT").replace(
        "stac-auth-proxy:8000", "localhost:8084"
    )
    browser_url = f"https://radiantearth.github.io/stac-browser/#/external/{browser_stac_url}/collections/{my_collection.id}"

IFrame(
    browser_url,
    1200,
    800,
)

2.3 Delete an item¶

It is less common but you might want to delete a STAC record some day. There are not yet any pypgstac functions for doing this but it is relatively easy to do via PostgreSQL query.

Try deleting the last item in the list.

with db.connect() as conn:
    cur = conn.cursor()
    cur.execute(f"DELETE FROM items where id = '{items[-1].id}';")
    cur.close()
    conn.commit()

new_n_items = db.query_one(
    f"SELECT COUNT(*) FROM items where collection = '{my_collection.id}';"
)
print(f"now there are {new_n_items} items")

# put it back :)
loader.load_items([items[-1].to_dict()], insert_mode=Methods.insert_ignore)

2.4 Search with pypgstac¶

It is not going to be the best option for searching a pgstac database in most cases (you should probably send a request to stac-fastapi-pgstac), but you can use pypgstac to search for items. You can provide a JSON request body like you would provide to a STAC API in the query arg of PgstacDB.search:

search_results = db.search(query={"collections": [my_collection.id], "limit": 1})

print(search_results)

2.5 Aggregations¶

Since your metadata is now cataloged in a PostgreSQL database, you can summarize it using SQL!

summary_query = db.query(
    "SELECT collection, COUNT(*) as count FROM items GROUP BY collection;"
)
for res in summary_query:
    print(res)

db.close()