Skip to content

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 
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
@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 
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
@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 
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
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 
 95
 96
 97
 98
 99
100
101
102
103
104
105
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

published_stac_item(flow_env, item, collection_name) async

" Push a STAC item into the rs-catalog.

Source code in docs/rs-client-libraries/rs_workflows/utils/catalog.py 
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
@task
async def published_stac_item(flow_env: FlowEnv, item: Item, collection_name: str) -> None:
    """ "
    Push a STAC item into the rs-catalog.
    """
    logger = get_run_logger()
    logger.info(f"The STAC item 🧊 '{item.id}' will be published on the collection '{collection_name}'.")
    items_metadata: list[DprProcessedItemMetadata] = []
    publish_mapping: list[FlowGeneratedProduct] = []

    items_metadata.append(
        DprProcessedItemMetadata(
            output_product_id=item.id,
            product_type=item.properties.get("product:type"),
            stac_item=item,
        ),
    )

    publish_mapping.append(
        FlowGeneratedProduct(
            name=item.id,
            product_type=str(item.properties.get("product:type")),
            collection_name=collection_name,
        ),
    )

    logger.debug(f"items_metadata = {items_metadata}")
    logger.debug(f"publish_mapping = {publish_mapping}")
    await publish(
        flow_env.serialize(),
        publish_mapping,
        items_metadata,
    )