API

The section shows information about the API relevant for the developers.

eodm.extract

extract_stac_api_collections(url)

Extracts collections from a STAC API

Parameters:

Name	Type	Description	Default
url	str	Link to STAC API endpoint	required

Yields:

Type	Description
Collection	Iterator[Collection]: pystac Collections

Source code in src/eodm/extract.py

def extract_stac_api_collections(url: str) -> Iterator[Collection]:
    """Extracts collections from a STAC API

    Args:
        url (str): Link to STAC API endpoint

    Yields:
        Iterator[Collection]: pystac Collections
    """

    client = pystac_client.Client.open(url)
    yield from client.get_collections()

extract_stac_api_items(url, collections=None, bbox=None, datetime_interval=None, limit=None)

Extracts items from a STAC API

Parameters:

Name	Type	Description	Default
url	str	Link to STAC API endpoint	required
collections	list[str] | None	List of collections to extract items from. Defaults to None.	None
bbox	tuple[float, float, float, float] | None	Bounding box to search. Defaults to None.	None
datetime_interval	str | None	Datetime interval to search. Defaults to None.	None
limit	int	Limit query to given number. Defaults to 10.	None

Yields:

Type	Description
Item	Iterator[Item]: pystac Items

Source code in src/eodm/extract.py

def extract_stac_api_items(
    url: str,
    collections: list[str] | None = None,
    bbox: tuple[float, float, float, float] | None = None,
    datetime_interval: str | None = None,
    limit: int | None = None,
) -> Iterator[Item]:
    """Extracts items from a STAC API

    Args:
        url (str): Link to STAC API endpoint
        collections (list[str] | None, optional): List of collections to extract items from. Defaults to None.
        bbox (tuple[float, float, float, float] | None, optional): Bounding box to search. Defaults to None.
        datetime_interval (str | None, optional): Datetime interval to search. Defaults to None.
        limit (int, optional): Limit query to given number. Defaults to 10.

    Yields:
        Iterator[Item]: pystac Items
    """

    client = pystac_client.Client.open(url)

    search = client.search(
        collections=collections,
        bbox=bbox,
        datetime=datetime_interval,
        limit=limit,
    )

    yield from search.item_collection()

eodm.load

DEFAULT_HEADERS = {'Content-Type': 'application/json'} module-attribute

load_stac_api_collections(url, collections, headers=None, verify=True, update=False, skip_existing=False)

Load multiple collections to a stac API

Parameters:

Name	Type	Description	Default
url	str	STAC API URL	required
collections	Iterable[Collection]	A collection of STAC Collections	required
headers	dict[str, str] | None	Additional headers to send. Defaults to None.	None
verify	bool	Verify TLS request. Defaults to True.	True
update	bool	Update the destination Collections. Defaults to False.	False
skip_existing	bool	Skip existing Collections. Defaults to False.	False

Returns:

Type	Description
Iterable[Collection]	Iterable[Collection]:

Source code in src/eodm/load.py

def load_stac_api_collections(
    url: str,
    collections: Iterable[Collection],
    headers: dict[str, str] | None = None,
    verify: bool = True,
    update: bool = False,
    skip_existing: bool = False,
) -> Iterable[Collection]:
    """Load multiple collections to a stac API

    Args:
        url (str): STAC API URL
        collections (Iterable[Collection]): A collection of STAC Collections
        headers (dict[str, str] | None, optional): Additional headers to send. Defaults to None.
        verify (bool, optional): Verify TLS request. Defaults to True.
        update (bool, optional): Update the destination Collections. Defaults to False.
        skip_existing (bool, optional): Skip existing Collections. Defaults to False.

    Returns:
        Iterable[Collection]:
    """

    if not headers:
        headers = DEFAULT_HEADERS

    collections_endpoint = f"{url}/collections"
    for collection in collections:
        response = httpx.post(
            collections_endpoint,
            json=collection.to_dict(),
            headers=headers,
            verify=verify,
        )
        if response.status_code == 409:
            if update:
                collection_endpoint = f"{collections_endpoint}/{collection.id}"
                response = httpx.put(
                    collection_endpoint,
                    json=collection.to_dict(),
                    headers=headers,
                    verify=verify,
                )
            if skip_existing:
                continue

        response.raise_for_status()
        yield collection

load_stac_api_items(url, items, headers=None, verify=True, update=False, skip_existing=False)

Load multiple items into a STAC API

Parameters:

Name	Type	Description	Default
url	str	STAC API url	required
items	Iterable[Item]	A collection of STAC Items	required
headers	dict[str, str] | None	Headers to add to the request. Defaults to None.	None
verify	bool	Verify SSL request. Defaults to True.	True
update	bool	Update STAC Item with new content. Defaults to False.	False
skip_existing	bool	Skip Item if exists. Defaults to False.	False

Source code in src/eodm/load.py

def load_stac_api_items(
    url: str,
    items: Iterable[Item],
    headers: dict[str, str] | None = None,
    verify: bool = True,
    update: bool = False,
    skip_existing: bool = False,
) -> Iterable[Item]:
    """Load multiple items into a STAC API

    Args:
        url (str): STAC API url
        items (Iterable[Item]): A collection of STAC Items
        headers (dict[str, str] | None, optional): Headers to add to the request. Defaults to None.
        verify (bool, optional): Verify SSL request. Defaults to True.
        update (bool, optional): Update STAC Item with new content. Defaults to False.
        skip_existing (bool, optional): Skip Item if exists. Defaults to False.
    """
    if not headers:
        headers = DEFAULT_HEADERS

    for item in items:
        collection_id = item.collection_id
        items_endpoint = f"{url}/collections/{collection_id}/items"
        response = httpx.post(
            items_endpoint,
            json=item.to_dict(),
            headers=headers,
            verify=verify,
        )
        if response.status_code == 409:
            if update:
                item_endpoint = f"{items_endpoint}/{item.id}"
                response = httpx.put(
                    item_endpoint, json=item.to_dict(), headers=headers, verify=verify
                )
            if skip_existing:
                continue
        response.raise_for_status()
        yield item

eodm.serializers

Mappable

Bases: Protocol

to_dict()

Source code in src/eodm/serializers.py

def to_dict(self): ...

default_serializer(items)

Serializes a list of Mappables (implementing to_dict()) to json strings individually

Parameters:

Name	Type	Description	Default
items	Iterable[Mappable]	A collection of Mappable items	required

Returns:

Type	Description
Iterable[str]	Iterable[str]: item as a string

Yields:

Type	Description
Iterable[str]	Iterator[Iterable[str]]: Collection of json strings

Source code in src/eodm/serializers.py

def default_serializer(items: Iterable[Mappable]) -> Iterable[str]:
    """Serializes a list of Mappables (implementing to_dict()) to json strings
    individually


    Args:
        items (Iterable[Mappable]): A collection of Mappable items

    Returns:
        Iterable[str]: item as a string

    Yields:
        Iterator[Iterable[str]]: Collection of json strings
    """

    for item in items:
        yield json.dumps(item.to_dict())

json_serializer(items)

Serializes a list of Mappables (implementing to_dict()) to a json list

Parameters:

Name	Type	Description	Default
items	Iterable[Mappable]	A collection of Mappable items	required

Returns:

Name	Type	Description
str	str	items as a json list

Source code in src/eodm/serializers.py

def json_serializer(items: Iterable[Mappable]) -> str:
    """Serializes a list of Mappables (implementing to_dict()) to a json list

    Args:
        items (Iterable[Mappable]): A collection of Mappable items

    Returns:
        str: items as a json list
    """
    return json.dumps([item.to_dict() for item in items])

eodm.stac_contrib

__all__ = ['FSSpecStacIO'] module-attribute

FSSpecStacIO

Bases: StacIO

Extension of StacIO to allow working with different filesystems in STAC using fsspec.

More information: https://pystac.readthedocs.io/en/stable/concepts.html#i-o-in-pystac

read_text(source, *args, **kwargs)

Source code in src/eodm/stac_contrib.py

def read_text(self, source: HREF, *args: Any, **kwargs: Any) -> str:
    if fs := kwargs.get("filesystem"):
        with fs.open(source, "r", *args, **kwargs) as f:
            return f.read()
    else:
        with fsspec.open(source, "r", *args, **kwargs) as f:
            return f.read()

write_text(dest, txt, *args, **kwargs)

Source code in src/eodm/stac_contrib.py

def write_text(self, dest: HREF, txt: str, *args: Any, **kwargs: Any) -> None:
    if fs := kwargs.get("filesystem"):
        with fs.open(dest, "w", *args, **kwargs) as f:
            f.write(txt)
    else:
        with fsspec.open(dest, "w", *args, **kwargs) as f:
            f.write(txt)