Stac

STAC stuff.

LogException

Bases: Exception

Inconsistent Asset Naming exception.

Source code in teledetection/upload/stac.py

class LogException(Exception):
    """Inconsistent Asset Naming exception."""

STACObjectUnresolved

Bases: Exception

Unresolved STAC object exception.

Source code in teledetection/upload/stac.py

class STACObjectUnresolved(Exception):
    """Unresolved STAC object exception."""

StacTransactionsHandler dataclass

Handle STAC and storage transactions.

Source code in teledetection/upload/stac.py

@dataclass
class StacTransactionsHandler:
    """Handle STAC and storage transactions."""

    stac_endpoint: str = DEFAULT_STAC_EP
    sign: bool = False
    """Sign URLs when fetching remote items."""

    @property
    def client(self):
        """STAC API client."""
        return pystac_client.Client.open(
            self.stac_endpoint,
            modifier=sign_inplace if self.sign else None,
        )

    def delete_item_or_col(self, col_id: str, item_id: str = ""):
        """Delete an item or a collection."""
        logger.info("Deleting %s%s", col_id, f"/{item_id}" if item_id else "")
        if item_id:
            url = f"{self.stac_endpoint}/collections/{col_id}/items/{item_id}"
        else:
            url = f"{self.stac_endpoint}/collections/{col_id}"
        resp = requests.delete(
            url,
            headers=get_headers(),
            timeout=TIMEOUT,
        )
        if resp.status_code != 200:
            logger.warning("Deletion failed (%s)", resp.text)

    def get_items(self, col_id: str, max_items: int = 10):
        """Get items in a collection."""
        logger.info("Get collection %s items", col_id)
        res = self.client.search(collections=[col_id], max_items=max_items)
        return list(res.items())

    def get_item(self, col_id: str, item_id: str) -> Item:
        """Retrieve a remote item."""
        logger.info("Retrieve item %s from collection %s", item_id, col_id)
        col = self.client.get_collection(col_id)
        if not col:
            raise KeyError(f"Collection {col_id} not found")
        item = col.get_item(item_id)
        if not item:
            raise UnconsistentCollectionIDs(
                f"Item {item_id} (from collection {col_id}) not found"
            )
        return item

    def publish_collection(self, col: Collection):
        """Publish an empty collection."""
        _check_naming_is_compliant(col.id)
        logger.info('Publishing collection "%s"', col.id)
        post_or_put(url=urljoin(self.stac_endpoint, "/collections"), data=col.to_dict())

    def publish_item(self, item: Item):
        """Publish an item."""
        _check_naming_is_compliant(item.id)
        col_id = item.collection_id
        logger.info('Publishing item "%s" in collection "%s"', item.id, col_id)
        post_or_put(
            urljoin(self.stac_endpoint, f"collections/{col_id}/items"),
            item.to_dict(transform_hrefs=False),
        )

    def publish_bulk_items(
        self, items: list[Item], method: str = "upsert", chunk_size: int = 500
    ):
        """Publish multiple items at once.

        Uses endpoint "/collections/{collection_id}/bulk_items" to publish chunks
        of multiple items, in order to reduce the number of requests.

        Args:
            items: List of items to publish
            method: "insert" or "upsert".
                    When an item already exists:
                        - if "insert", an error rises.
                        - if "upsert", the item is updated.
                    In both cases, the item is created if it does not exist.
            chunk_size: maximum number of items per request.

        Returns:
            None

        Notes:
            `chunk_size` allows to limit the number of items per request in order to
            respect the response timeout. Otherwise, it is possible to control the
            timeout with the `TIMEOUT` environment variable or
            `teledetection.sdk.settings.ENV.tld_request_timeout`.
        """
        for item in items:
            _check_naming_is_compliant(item.id)
        if len(set(item.collection_id for item in items)) > 1:
            raise UnconsistentCollectionIDs(
                "Collection ID must be the same for all items!"
            )
        if len(items) == 0:
            return

        col_id = items[0].collection_id
        bulk_items = {
            item.id: item.to_dict(transform_hrefs=False) for item in items[:chunk_size]
        }
        logger.info('Publishing %d items in collection "%s"', len(bulk_items), col_id)
        post_or_put(
            url=urljoin(self.stac_endpoint, f"collections/{col_id}/bulk_items"),
            data={
                "method": method,
                "items": bulk_items,
            },
        )
        logger.info(f"Published {len(bulk_items)} items")

        if len(items) > chunk_size:
            self.publish_bulk_items(items=items[chunk_size:])

    def update_collection_extent(self, col_id: str):
        """Update collection extent."""
        logger.info("Updating collection extent")
        results = self.client.search(limit=1000, collections=[col_id])
        col = self.client.get_collection(col_id)
        col.extent = pystac.Extent.from_items(items=list(results.items()))
        col.clear_links("items")
        self.publish_collection(col=col)

    def load_and_save(
        self, col_id: str, obj_pth: str, item_id: str = "", pretty: bool = True
    ):
        """Load and save locally (as .json file) the remote STAC object."""
        obj = (
            self.get_item(col_id=col_id, item_id=item_id)
            if item_id
            else self.client.get_collection(col_id)
        )
        logger.info("Writing file %s", obj_pth)
        with open(obj_pth, "w", encoding="utf-8") as file:
            json.dump(obj.to_dict(), file, indent=2 if pretty else None)

    def load_and_publish(self, obj_pth: str):
        """Load and publish the serialized STAC object."""
        obj = load_stac_obj(obj_pth=obj_pth)
        if isinstance(obj, Item):
            self.publish_item(item=obj)
        elif isinstance(obj, Collection):
            self.publish_collection(col=obj)
        elif isinstance(obj, ItemCollection):
            for item in obj.items:
                self.publish_item(item=item)
        else:
            raise TypeError(
                f"Invalid type, must be ItemCollection or Collection (got {type(obj)})"
            )

client property

STAC API client.

sign = False class-attribute instance-attribute

Sign URLs when fetching remote items.

delete_item_or_col(col_id, item_id='')

Delete an item or a collection.

Source code in teledetection/upload/stac.py

def delete_item_or_col(self, col_id: str, item_id: str = ""):
    """Delete an item or a collection."""
    logger.info("Deleting %s%s", col_id, f"/{item_id}" if item_id else "")
    if item_id:
        url = f"{self.stac_endpoint}/collections/{col_id}/items/{item_id}"
    else:
        url = f"{self.stac_endpoint}/collections/{col_id}"
    resp = requests.delete(
        url,
        headers=get_headers(),
        timeout=TIMEOUT,
    )
    if resp.status_code != 200:
        logger.warning("Deletion failed (%s)", resp.text)

get_item(col_id, item_id)

Retrieve a remote item.

Source code in teledetection/upload/stac.py

def get_item(self, col_id: str, item_id: str) -> Item:
    """Retrieve a remote item."""
    logger.info("Retrieve item %s from collection %s", item_id, col_id)
    col = self.client.get_collection(col_id)
    if not col:
        raise KeyError(f"Collection {col_id} not found")
    item = col.get_item(item_id)
    if not item:
        raise UnconsistentCollectionIDs(
            f"Item {item_id} (from collection {col_id}) not found"
        )
    return item

get_items(col_id, max_items=10)

Get items in a collection.

Source code in teledetection/upload/stac.py

def get_items(self, col_id: str, max_items: int = 10):
    """Get items in a collection."""
    logger.info("Get collection %s items", col_id)
    res = self.client.search(collections=[col_id], max_items=max_items)
    return list(res.items())

load_and_publish(obj_pth)

Load and publish the serialized STAC object.

Source code in teledetection/upload/stac.py

def load_and_publish(self, obj_pth: str):
    """Load and publish the serialized STAC object."""
    obj = load_stac_obj(obj_pth=obj_pth)
    if isinstance(obj, Item):
        self.publish_item(item=obj)
    elif isinstance(obj, Collection):
        self.publish_collection(col=obj)
    elif isinstance(obj, ItemCollection):
        for item in obj.items:
            self.publish_item(item=item)
    else:
        raise TypeError(
            f"Invalid type, must be ItemCollection or Collection (got {type(obj)})"
        )

load_and_save(col_id, obj_pth, item_id='', pretty=True)

Load and save locally (as .json file) the remote STAC object.

Source code in teledetection/upload/stac.py

def load_and_save(
    self, col_id: str, obj_pth: str, item_id: str = "", pretty: bool = True
):
    """Load and save locally (as .json file) the remote STAC object."""
    obj = (
        self.get_item(col_id=col_id, item_id=item_id)
        if item_id
        else self.client.get_collection(col_id)
    )
    logger.info("Writing file %s", obj_pth)
    with open(obj_pth, "w", encoding="utf-8") as file:
        json.dump(obj.to_dict(), file, indent=2 if pretty else None)

publish_bulk_items(items, method='upsert', chunk_size=500)

Publish multiple items at once.

Uses endpoint "/collections/{collection_id}/bulk_items" to publish chunks of multiple items, in order to reduce the number of requests.

Parameters:

Name	Type	Description	Default
items	list[Item]	List of items to publish	required
method	str	"insert" or "upsert". When an item already exists: - if "insert", an error rises. - if "upsert", the item is updated. In both cases, the item is created if it does not exist.	'upsert'
chunk_size	int	maximum number of items per request.	500

Returns:

Type	Description
	None

Notes

chunk_size allows to limit the number of items per request in order to respect the response timeout. Otherwise, it is possible to control the timeout with the TIMEOUT environment variable or teledetection.sdk.settings.ENV.tld_request_timeout.

Source code in teledetection/upload/stac.py

def publish_bulk_items(
    self, items: list[Item], method: str = "upsert", chunk_size: int = 500
):
    """Publish multiple items at once.

    Uses endpoint "/collections/{collection_id}/bulk_items" to publish chunks
    of multiple items, in order to reduce the number of requests.

    Args:
        items: List of items to publish
        method: "insert" or "upsert".
                When an item already exists:
                    - if "insert", an error rises.
                    - if "upsert", the item is updated.
                In both cases, the item is created if it does not exist.
        chunk_size: maximum number of items per request.

    Returns:
        None

    Notes:
        `chunk_size` allows to limit the number of items per request in order to
        respect the response timeout. Otherwise, it is possible to control the
        timeout with the `TIMEOUT` environment variable or
        `teledetection.sdk.settings.ENV.tld_request_timeout`.
    """
    for item in items:
        _check_naming_is_compliant(item.id)
    if len(set(item.collection_id for item in items)) > 1:
        raise UnconsistentCollectionIDs(
            "Collection ID must be the same for all items!"
        )
    if len(items) == 0:
        return

    col_id = items[0].collection_id
    bulk_items = {
        item.id: item.to_dict(transform_hrefs=False) for item in items[:chunk_size]
    }
    logger.info('Publishing %d items in collection "%s"', len(bulk_items), col_id)
    post_or_put(
        url=urljoin(self.stac_endpoint, f"collections/{col_id}/bulk_items"),
        data={
            "method": method,
            "items": bulk_items,
        },
    )
    logger.info(f"Published {len(bulk_items)} items")

    if len(items) > chunk_size:
        self.publish_bulk_items(items=items[chunk_size:])

publish_collection(col)

Publish an empty collection.

Source code in teledetection/upload/stac.py

def publish_collection(self, col: Collection):
    """Publish an empty collection."""
    _check_naming_is_compliant(col.id)
    logger.info('Publishing collection "%s"', col.id)
    post_or_put(url=urljoin(self.stac_endpoint, "/collections"), data=col.to_dict())

publish_item(item)

Publish an item.

Source code in teledetection/upload/stac.py

def publish_item(self, item: Item):
    """Publish an item."""
    _check_naming_is_compliant(item.id)
    col_id = item.collection_id
    logger.info('Publishing item "%s" in collection "%s"', item.id, col_id)
    post_or_put(
        urljoin(self.stac_endpoint, f"collections/{col_id}/items"),
        item.to_dict(transform_hrefs=False),
    )

update_collection_extent(col_id)

Update collection extent.

Source code in teledetection/upload/stac.py

def update_collection_extent(self, col_id: str):
    """Update collection extent."""
    logger.info("Updating collection extent")
    results = self.client.search(limit=1000, collections=[col_id])
    col = self.client.get_collection(col_id)
    col.extent = pystac.Extent.from_items(items=list(results.items()))
    col.clear_links("items")
    self.publish_collection(col=col)

StacUploadTransactionsHandler dataclass

Bases: StacTransactionsHandler

Handle STAC and storage transactions.

Source code in teledetection/upload/stac.py

@dataclass
class StacUploadTransactionsHandler(StacTransactionsHandler):
    """Handle STAC and storage transactions."""

    storage_endpoint: str = DEFAULT_S3_EP
    storage_bucket: str = DEFAULT_S3_STORAGE
    assets_overwrite: bool = False
    """Overwrite assets on S3 if already existing."""
    keep_cog_dir: str = ""

    def publish_item_and_push_assets(self, item: Item, assets_root_dir: str):
        """Publish an item and push all its assets.

        Args:
            item: Stac item to publish
            assets_root_dir: Common path to all files, defined as assets root dir
        """
        logger.debug("Itemid = %s", item.id)

        _check_naming_is_compliant(self.storage_bucket)
        _check_naming_is_compliant(item.id)

        for _, asset in item.assets.items():
            assert item.collection_id
            self.push_asset_and_update_href(asset, assets_root_dir, item.collection_id)

        # Add published metadata to item
        logger.debug("Updating item metadata ...")
        raster.apply_created_metadata(item)

        # Push item
        self.publish_item(item=item)

    def publish_items_and_push_assets(self, items: list[Item]):
        """Publish items."""
        if not items:
            logger.info("No item to publish.")
            return
        check_items_col_id(items=items)
        assets_root_dir = get_assets_root_dir(items=items)
        logger.debug("Assets root directory: %s", assets_root_dir)
        for item in items:
            self.publish_item_and_push_assets(
                item=item, assets_root_dir=assets_root_dir
            )
        # Update collection extent
        col_id = items[0].collection_id
        if not col_id:
            raise UnconsistentCollectionIDs(
                f"Collection id is None for item {items[0].id}"
            )
        self.update_collection_extent(col_id=col_id)

    def push_asset_and_update_href(
        self, asset: pystac.Asset, assets_root_dir: str, col_id: str
    ):
        """Push an asset to the storage and update href and media_type."""
        tgt_root_url = urljoin(
            self.storage_endpoint, f"{self.storage_bucket}/{col_id}/"
        )
        local_filename = asset.href
        if local_filename.startswith(("https://", "http://")):
            logger.warning(f"{local_filename} is not local, asset will not be pushed")
            return
        logger.debug("Local file: %s", local_filename)

        file_relative_path = os.path.relpath(local_filename, assets_root_dir)
        target_url = urljoin(tgt_root_url, file_relative_path)

        # Check that url part after storage bucket is compliant
        _check_naming_is_compliant(
            file_relative_path,
            allow_dot=True,
            allow_slash=True,
        )
        logger.debug("Target file url: %s", target_url)

        # Add raster metadata to asset
        logger.debug("Updating assets metadata for rasters...")
        if raster.is_raster(local_filename):
            raster.apply_proj_extension(asset)
            raster.apply_raster_extension(asset)
            asset.media_type = pystac.MediaType.COG

        # Skip when target file exists and overwrite is not enabled
        if not self.assets_overwrite:
            if asset_exists(target_url):
                asset.href = target_url
                return

        # Check is_cog, converts if not
        cogconv = False
        if raster.is_raster(local_filename):
            if not raster.is_cog(local_filename):
                orig_filename = local_filename
                local_filename = raster.convert_to_cog(
                    orig_filename,
                    keep_cog_dir=self.keep_cog_dir,
                )
                cogconv = True

        # Upload file
        logger.info("Uploading %s to %s...", local_filename, target_url)
        try:
            push(local_filename=local_filename, target_url=target_url)
        except Exception as e:
            logger.error(e)
            raise e

        # Update assets hrefs
        logger.debug("Updating assets HREFs ...")
        asset.href = target_url

        # Delete temp cog
        if cogconv and not self.keep_cog_dir:
            logger.debug("Deleting temporary COG ...")
            shutil.rmtree(os.path.dirname(local_filename))
            local_filename = orig_filename

    def publish_collection_and_push_assets(self, col: Collection):
        """Publish a collection and push all its assets."""
        logger.debug("Collection = %s", col.id)

        _check_naming_is_compliant(self.storage_bucket)
        _check_naming_is_compliant(col.id)
        if len(col.assets) > 0:
            assets_root_dir = get_assets_root_dir(items=[], collection=col)
            for _, asset in col.assets.items():
                self.push_asset_and_update_href(asset, assets_root_dir, col.id)
        self.publish_collection(col=col)

    def publish_collection_with_items(self, col: Collection):
        """Publish a collection and all its items."""
        items = get_col_items(col=col)
        check_items_col_id(items)
        self.publish_collection_and_push_assets(col=col)
        self.publish_items_and_push_assets(items=items)

    def publish_item_collection(self, item_collection: ItemCollection):
        """Publish an item collection and all of its items."""
        self.publish_items_and_push_assets(items=item_collection.items)

    def load_and_publish(self, obj_pth: str):
        """Load and publish the serialized STAC object."""
        obj = load_stac_obj(obj_pth=obj_pth)
        if isinstance(obj, Item):
            self.publish_items_and_push_assets(items=[obj])
        elif isinstance(obj, Collection):
            self.publish_collection_with_items(col=obj)
        elif isinstance(obj, ItemCollection):
            self.publish_item_collection(item_collection=obj)

assets_overwrite = False class-attribute instance-attribute

Overwrite assets on S3 if already existing.

load_and_publish(obj_pth)

Load and publish the serialized STAC object.

Source code in teledetection/upload/stac.py

def load_and_publish(self, obj_pth: str):
    """Load and publish the serialized STAC object."""
    obj = load_stac_obj(obj_pth=obj_pth)
    if isinstance(obj, Item):
        self.publish_items_and_push_assets(items=[obj])
    elif isinstance(obj, Collection):
        self.publish_collection_with_items(col=obj)
    elif isinstance(obj, ItemCollection):
        self.publish_item_collection(item_collection=obj)

publish_collection_and_push_assets(col)

Publish a collection and push all its assets.

Source code in teledetection/upload/stac.py

def publish_collection_and_push_assets(self, col: Collection):
    """Publish a collection and push all its assets."""
    logger.debug("Collection = %s", col.id)

    _check_naming_is_compliant(self.storage_bucket)
    _check_naming_is_compliant(col.id)
    if len(col.assets) > 0:
        assets_root_dir = get_assets_root_dir(items=[], collection=col)
        for _, asset in col.assets.items():
            self.push_asset_and_update_href(asset, assets_root_dir, col.id)
    self.publish_collection(col=col)

publish_collection_with_items(col)

Publish a collection and all its items.

Source code in teledetection/upload/stac.py

def publish_collection_with_items(self, col: Collection):
    """Publish a collection and all its items."""
    items = get_col_items(col=col)
    check_items_col_id(items)
    self.publish_collection_and_push_assets(col=col)
    self.publish_items_and_push_assets(items=items)

publish_item_and_push_assets(item, assets_root_dir)

Publish an item and push all its assets.

Parameters:

Name	Type	Description	Default
item	Item	Stac item to publish	required
assets_root_dir	str	Common path to all files, defined as assets root dir	required

Source code in teledetection/upload/stac.py

def publish_item_and_push_assets(self, item: Item, assets_root_dir: str):
    """Publish an item and push all its assets.

    Args:
        item: Stac item to publish
        assets_root_dir: Common path to all files, defined as assets root dir
    """
    logger.debug("Itemid = %s", item.id)

    _check_naming_is_compliant(self.storage_bucket)
    _check_naming_is_compliant(item.id)

    for _, asset in item.assets.items():
        assert item.collection_id
        self.push_asset_and_update_href(asset, assets_root_dir, item.collection_id)

    # Add published metadata to item
    logger.debug("Updating item metadata ...")
    raster.apply_created_metadata(item)

    # Push item
    self.publish_item(item=item)

publish_item_collection(item_collection)

Publish an item collection and all of its items.

Source code in teledetection/upload/stac.py

def publish_item_collection(self, item_collection: ItemCollection):
    """Publish an item collection and all of its items."""
    self.publish_items_and_push_assets(items=item_collection.items)

publish_items_and_push_assets(items)

Publish items.

Source code in teledetection/upload/stac.py

def publish_items_and_push_assets(self, items: list[Item]):
    """Publish items."""
    if not items:
        logger.info("No item to publish.")
        return
    check_items_col_id(items=items)
    assets_root_dir = get_assets_root_dir(items=items)
    logger.debug("Assets root directory: %s", assets_root_dir)
    for item in items:
        self.publish_item_and_push_assets(
            item=item, assets_root_dir=assets_root_dir
        )
    # Update collection extent
    col_id = items[0].collection_id
    if not col_id:
        raise UnconsistentCollectionIDs(
            f"Collection id is None for item {items[0].id}"
        )
    self.update_collection_extent(col_id=col_id)

push_asset_and_update_href(asset, assets_root_dir, col_id)

Push an asset to the storage and update href and media_type.

Source code in teledetection/upload/stac.py

def push_asset_and_update_href(
    self, asset: pystac.Asset, assets_root_dir: str, col_id: str
):
    """Push an asset to the storage and update href and media_type."""
    tgt_root_url = urljoin(
        self.storage_endpoint, f"{self.storage_bucket}/{col_id}/"
    )
    local_filename = asset.href
    if local_filename.startswith(("https://", "http://")):
        logger.warning(f"{local_filename} is not local, asset will not be pushed")
        return
    logger.debug("Local file: %s", local_filename)

    file_relative_path = os.path.relpath(local_filename, assets_root_dir)
    target_url = urljoin(tgt_root_url, file_relative_path)

    # Check that url part after storage bucket is compliant
    _check_naming_is_compliant(
        file_relative_path,
        allow_dot=True,
        allow_slash=True,
    )
    logger.debug("Target file url: %s", target_url)

    # Add raster metadata to asset
    logger.debug("Updating assets metadata for rasters...")
    if raster.is_raster(local_filename):
        raster.apply_proj_extension(asset)
        raster.apply_raster_extension(asset)
        asset.media_type = pystac.MediaType.COG

    # Skip when target file exists and overwrite is not enabled
    if not self.assets_overwrite:
        if asset_exists(target_url):
            asset.href = target_url
            return

    # Check is_cog, converts if not
    cogconv = False
    if raster.is_raster(local_filename):
        if not raster.is_cog(local_filename):
            orig_filename = local_filename
            local_filename = raster.convert_to_cog(
                orig_filename,
                keep_cog_dir=self.keep_cog_dir,
            )
            cogconv = True

    # Upload file
    logger.info("Uploading %s to %s...", local_filename, target_url)
    try:
        push(local_filename=local_filename, target_url=target_url)
    except Exception as e:
        logger.error(e)
        raise e

    # Update assets hrefs
    logger.debug("Updating assets HREFs ...")
    asset.href = target_url

    # Delete temp cog
    if cogconv and not self.keep_cog_dir:
        logger.debug("Deleting temporary COG ...")
        shutil.rmtree(os.path.dirname(local_filename))
        local_filename = orig_filename

UnconsistentAssetNaming

Bases: Exception

Inconsistent Asset Naming exception.

Source code in teledetection/upload/stac.py

class UnconsistentAssetNaming(Exception):
    """Inconsistent Asset Naming exception."""

UnconsistentCollectionIDs

Bases: Exception

Inconsistent STAC collection exception.

Source code in teledetection/upload/stac.py

class UnconsistentCollectionIDs(Exception):
    """Inconsistent STAC collection exception."""

asset_exists(asset_url)

Check that the item provided in parameter exists and is accessible.

Source code in teledetection/upload/stac.py

def asset_exists(asset_url: str) -> bool:
    """Check that the item provided in parameter exists and is accessible."""
    sess = create_session()
    asset_url_signed = sign(asset_url)
    res = sess.get(asset_url_signed, stream=True)
    if res.status_code == 200:
        logger.info("Asset %s already exists.", asset_url)
        return True
    return False

check_items_col_id(items)

Check that items have the same col_id.

Source code in teledetection/upload/stac.py

def check_items_col_id(items: list[Item]):
    """Check that items have the same col_id."""
    if len(set(item.collection_id for item in items)) > 1:
        raise UnconsistentCollectionIDs("Collection ID must be the same for all items!")

create_session()

Create a requests session.

Source code in teledetection/upload/stac.py

def create_session():
    """Create a requests session."""
    sess = requests.Session()
    retries = Retry(
        total=3,
        backoff_factor=1,
        status_forcelist=[
            408,
            419,
            425,
            500,
            502,
            503,
            504,
        ],
        allowed_methods=frozenset(["PUT", "POST"]),
        raise_on_status=False,
    )
    adapter = HTTPAdapter(max_retries=retries)
    sess.mount("http://", adapter=adapter)
    sess.mount("https://", adapter=adapter)
    return sess

get_assets_root_dir(items, collection=None)

Get the common prefix of all items assets paths.

If the the common prefix is not a folder (/tmp/test1/a.tif, /tmp/test2/b.tif), returns /tmp.

Source code in teledetection/upload/stac.py

def get_assets_root_dir(
    items: list[Item], collection: Optional[Collection] = None
) -> str:
    """Get the common prefix of all items assets paths.

    If the the common prefix is not a folder (/tmp/test1/a.tif, /tmp/test2/b.tif), returns /tmp.
    """
    hrefs = [asset.href for item in items for asset in item.assets.values()]
    if collection:
        hrefs += [asset.href for asset in collection.assets.values()]
    prefix = os.path.commonpath(hrefs)
    if os.path.isdir(prefix):
        return prefix + "/"
    return os.path.dirname(prefix) + "/"

get_col_href(col)

Retrieve collection href.

Source code in teledetection/upload/stac.py

def get_col_href(col: Collection):
    """Retrieve collection href."""
    for link in col.links:
        if link.rel == "self":
            return link.href
    return ""

get_col_items(col)

Retrieve collection items.

Source code in teledetection/upload/stac.py

def get_col_items(col: Collection) -> list[Item]:
    """Retrieve collection items."""
    col_href = get_col_href(col=col)
    return [
        cast(
            Item,
            load_stac_obj(
                os.path.join(os.path.dirname(col_href), link.href[2:])
                if link.href.startswith("./")
                else link.href
            ),
        )
        for link in col.links
        if link.rel == "item"
    ]

load_stac_obj(obj_pth)

Load a STAC object serialized on disk.

Source code in teledetection/upload/stac.py

def load_stac_obj(obj_pth: str) -> Collection | ItemCollection | Item:
    """Load a STAC object serialized on disk."""
    for obj_name, cls in {
        "collection": Collection,
        "item collection": ItemCollection,
        "item": Item,
    }.items():
        logger.debug("Try to read file %s", obj_pth)
        try:
            obj = getattr(cls, "from_file")(obj_pth)
            logger.info("Loaded %s from file %s", obj_name, obj_pth)
            logger.debug(obj.to_dict())
            return obj
        except pystac.errors.STACTypeError:
            pass

    raise STACObjectUnresolved(f"Cannot resolve STAC object ({obj_pth})")

post_or_put(url, data)

Post or put data to url.

Source code in teledetection/upload/stac.py

def post_or_put(url: str, data: dict):
    """Post or put data to url."""
    headers = get_headers()
    sess = create_session()

    resp = sess.post(url, json=data, headers=headers, timeout=TIMEOUT)

    if resp.status_code == 409:
        # Exists, so update
        logger.info("Item at %s already exists, doing a PUT", url)
        resp = sess.put(
            f"{url}/{data['id']}",
            json=data,
            headers=headers,
            timeout=TIMEOUT,
        )
        # Unchanged may throw a 404
        if not resp.status_code == 404:
            resp.raise_for_status()

    try:
        resp.raise_for_status()
    except HTTPError as e:
        try:
            logger.error("Server returned: %s", pretty_repr(resp.json()))
        except LogException:
            logger.error("Server returned: %s", resp.text)
        raise e