rs_workflows/utils/catalog.md

<< Back to index

Helper task to interact with the rs-catalog.

get_catalog_items(flow_env, item_ids, collections) async

Get items from a set of rs-catalog collections

Source code in docs/rs-client-libraries/rs_workflows/utils/catalog.py

@task(name="Retrieve rs-catalog items from collections")
async def get_catalog_items(flow_env: FlowEnv, item_ids: list[str], collections: list[str]) -> ItemCollection | None:
    """
    Get items from a set of rs-catalog collections
    """
    get_run_logger().info(
        f"Search items {', '.join(item_ids)} in the collections {', '.join(collections)} from the rs-catalog.",
    )
    size = len(item_ids)
    return flow_env.rs_client.get_catalog_client().search(
        method="POST",
        collections=collections,
        ids=item_ids,
        max_items=size,
        limit=size,
    )

get_single_catalog_item(flow_env, item_id, collections) async

Get an item from a set of rs-catalog collections

Source code in docs/rs-client-libraries/rs_workflows/utils/catalog.py

@task(name="Retrieve rs-catalog item from collection")
async def get_single_catalog_item(flow_env: FlowEnv, item_id: str, collections: list[str]) -> Item | None:
    """
    Get an item from a set of rs-catalog collections
    """
    logger = get_run_logger()
    result: Item | None = None

    # Try to retrieve the item in the collections
    item_collection: ItemCollection | None = await get_catalog_items(flow_env, [item_id], collections)

    count = 0
    if item_collection is not None:
        count = len(item_collection.items)
    if count == 1:
        # One item was found on the rs-catalog
        logger.info(
            f"✅ The STAC item 🧊 '{item_id}' has been found on the rs-catalog collections {', '.join(collections)}.",
        )
        if item_collection is not None:
            result = item_collection.items[0]
    else:
        logger.warning(
            f"❌ The STAC item 🧊 '{item_id}' was not found on the rs-catalog collections {', '.join(collections)}.",
        )

    return result

is_evicted(item)

Check if the item is evicted.

Source code in docs/rs-client-libraries/rs_workflows/utils/catalog.py

def is_evicted(item: Item) -> tuple[bool, datetime | None]:
    """
    Check if the item is evicted.
    """
    eviction_date_str: str = ""

    for asset in item.assets.values():
        if "eviction_datetime" in asset.extra_fields:
            eviction_date_str = asset.extra_fields["eviction_datetime"]
            break

    if eviction_date_str:
        eviction_date = datetime.fromisoformat(eviction_date_str.replace("Z", "+00:00"))
        return eviction_date <= datetime.now(timezone.utc), eviction_date

    return False, None

is_published(item)

Check if the item is published.

Source code in docs/rs-client-libraries/rs_workflows/utils/catalog.py

def is_published(item: Item) -> bool:
    """
    Check if the item is published.
    """
    published_date_str = item.properties.get("published")

    if published_date_str:
        published_date = datetime.fromisoformat(published_date_str.replace("Z", "+00:00"))
        return published_date <= datetime.now(timezone.utc)

    return False