rs_workflows/utils/catalog.md

<< Back to index

Helper task to interact with the rs-catalog.

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 session on the collection
    catalog_client: CatalogClient = flow_env.rs_client.get_catalog_client()
    logger.info(f"Search item {item_id} on the collections {', '.join(collections)} from the  rs-catalog.")
    item_collection: ItemCollection | None = catalog_client.search(
        method="POST",
        collections=collections,
        ids=[item_id],
        limit=1,
    )

    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