rs_workflows/catalog_flow.md

<< Back to index

Catalog flow implementation

catalog_search(env, catalog_cql2, error_if_empty=False, collections=None) async

Search Catalog items.

Parameters:

Name	Type	Description	Default
env	FlowEnvArgs	Prefect flow environment (at least the owner_id is required)	required
catalog_cql2	dict	CQL2 filter.	required
error_if_empty	bool	Raise a ValueError if the results are empty.	False
collections	list[str] | None	list of collection names to search in	None

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

@flow(name="search-catalog")
async def catalog_search(
    env: FlowEnvArgs,
    catalog_cql2: dict,
    error_if_empty: bool = False,
    collections: list[str] | None = None,
) -> ItemCollection | None:
    """
    Search Catalog items.

    Args:
        env: Prefect flow environment (at least the owner_id is required)
        catalog_cql2: CQL2 filter.
        error_if_empty: Raise a ValueError if the results are empty.
        collections: list of collection names to search in
    """
    logger = get_logger()

    # Init flow environment and opentelemetry span
    flow_env = FlowEnv(env)
    with flow_env.start_span(__name__, "catalog-search"):

        logger.info(f"Start Catalog search: {catalog_cql2}")
        catalog_client: CatalogClient = flow_env.rs_client.get_catalog_client()
        found = catalog_client.search(
            method="POST",
            stac_filter=catalog_cql2.get("filter"),
            max_items=catalog_cql2.get("limit"),
            sortby=catalog_cql2.get("sortby"),
            collections=collections,
        )
        if (not found) and error_if_empty:
            raise ValueError(
                f"No Catalog item found for CQL2 filter: {json.dumps(catalog_cql2, indent=2)}",
            )
        logger.info(f"Catalog search found {len(found)} results: {found}")  # type: ignore
        return found

catalog_search_task(*args, **kwargs) async

See: search

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

@task(name="search-catalog")
async def catalog_search_task(*args, **kwargs) -> ItemCollection | None:
    """See: search"""
    return await catalog_search.fn(*args, **kwargs)

check_and_create_collection(flow_env, collection_name) async

Check if a collection exists, and create it if it doesn't.

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

@task(name="Check and create the collection if needed")
async def check_and_create_collection(flow_env: FlowEnv, collection_name: str):
    """
    Check if a collection exists, and create it if it doesn't.
    """
    # Check that the collection "collection_name" exists. Otherwise create it.
    logger = get_logger()

    catalog_client: CatalogClient = flow_env.rs_client.get_catalog_client()
    try:
        catalog_client.search(collections=[collection_name])
    except RuntimeError:
        # The collection is missing, we will create it
        logger.info(f"The collection {collection_name} is missing; it will be created.")
        spatial = SpatialExtent(bboxes=[[-94.6911621, 37.0332547, -94.402771, 37.1077651]])
        date_strings = ["2000-02-01T00:00:00Z", "2100-02-12T00:00:00Z"]
        date_objects: list[datetime | None] = [
            datetime.strptime(date_str, "%Y-%m-%dT%H:%M:%SZ") for date_str in date_strings
        ]
        temporal = TemporalExtent(intervals=date_objects)
        extent = Extent(spatial=spatial, temporal=temporal)
        new_collection = Collection(
            id=collection_name,
            description=f"{collection_name} collection",
            extent=extent,
        )
        catalog_client.add_collection(new_collection)

get_item(env, target_collection, item) async

Get a catalog item by its ID.

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

@task(name="Get catalog item")
async def get_item(
    env: FlowEnvArgs,
    target_collection,
    item,
):
    """
    Get a catalog item by its ID.
    """
    flow_env = FlowEnv(env)
    catalog_client: CatalogClient = flow_env.rs_client.get_catalog_client()
    return catalog_client.get_item(target_collection, item)

publish(env, generated_product_to_collection_identifier, items_metadata) async

Publish items to the catalog

Parameters:

Name	Type	Description	Default
env	FlowEnvArgs	Prefect flow environment	required
collection		Catalog collection identifier where the items are published	required
items_metadata	list[DprProcessedItemMetadata]	List of DprProcessedItemMetadata containing items to publish	required

Returns:

Type	Description
list[dict[str, Any]]	The successfully published STAC items serialized as JSON-compatible dictionaries.

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

@task(name="Publish to catalog")
async def publish(
    env: FlowEnvArgs,
    generated_product_to_collection_identifier: list[FlowGeneratedProduct],
    items_metadata: list[DprProcessedItemMetadata],
) -> list[dict[str, Any]]:
    """
    Publish items to the catalog

    Args:
        env: Prefect flow environment
        collection: Catalog collection identifier where the items are published
        items_metadata: List of DprProcessedItemMetadata containing items to publish

    Returns:
        The successfully published STAC items serialized as JSON-compatible dictionaries.
    """
    logger = get_logger()
    flow_env = FlowEnv(env)

    catalog_client: CatalogClient = flow_env.rs_client.get_catalog_client()
    published_items: list[dict[str, Any]] = []

    with flow_env.start_span(__name__, "publish-to-catalog"):
        for item_metadata in items_metadata:
            item = item_metadata.stac_item
            try:
                # Resolve destination collection using metadata from STAC item
                target_collection = resolve_collection(
                    item_metadata,
                    generated_product_to_collection_identifier,
                )
                # TEMPFIX: Items generated by DPR with [] as geometry
                # Also, remove some fields that s1 l0 processor is not compliant with
                # regarding the catalog ingestion
                # TEMPFIX: Items generated by DPR may contain empty geometry list
                if not item.geometry:
                    item.geometry = None
                fix_geojson_geometry(item)
                # TEMPFIX: Items generated by DPR may contain empty bbox list
                if not item.bbox:
                    item.bbox = None
                instruments = item.properties.get("instruments")
                if instruments is not None and not isinstance(instruments, list):
                    item.properties["instruments"] = [instruments]
                # TEMPFIX: Items generated by DPR may contain other
                # values than  ['producer', 'licensor', 'processor', 'host'] for ['properties']['providers']['roles']
                valid_roles = {"producer", "licensor", "processor", "host"}
                providers = item.properties.get("providers", [])
                valid_providers = []
                for provider in providers:
                    name = provider.get("name")
                    if not isinstance(name, str) or not name:
                        logger.warning("Provider name must be a nonempty string. Removing provider.")
                        continue
                    roles = provider.get("roles")
                    # ensure roles is a list before checking
                    if isinstance(roles, list):
                        # find first invalid role; returns None if all are valid
                        invalid_role = next((r for r in roles if r not in valid_roles), None)
                        if invalid_role:
                            logger.warning(
                                f"Provider role '{invalid_role}' is not valid. "
                                f"Valid roles: {valid_roles}. Removing provider.",
                            )
                            continue
                    valid_providers.append(provider)
                if item.properties.get("providers"):
                    item.properties["providers"] = valid_providers
                # TEMPFIX END

                # Add processing url link for each STAC item
                platform = os.environ.get("RSPY_PREFECT_URL")

                item.add_link(
                    Link(
                        rel="processing-execution",
                        media_type="application/json",
                        target=f"{platform}/runs/flow-run/{runtime.flow_run.id}",
                    ),
                )

                logger.info(
                    "Publishing product %s to %s. This may take a while...",
                    item.id,
                    target_collection,
                )

                # Check that the collection exists. Otherwise create it.
                await check_and_create_collection(flow_env, target_collection)

                # Publish item to catalog
                # TODO: adjust timeout as needed. Current value is 6 hours.
                logger.info(
                    "Using timeout of 21600 seconds for catalog item publishing",
                )
                try:
                    response = catalog_client.add_item(target_collection, item, timeout=21600)
                except Exception as e:  # pylint: disable=broad-exception-caught
                    # continue to the next item (cf request from story 1125)
                    logger.warning(
                        f"Exception `{e}` while publishing item: {json.dumps(item.to_dict(), indent=2)}",
                    )
                    continue
                logger.info(
                    "Publishing product %s to %s done. Response %s with message: \n%s",
                    item.id,
                    target_collection,
                    response.status_code,
                    response.text,
                )
                published_item = item.to_dict()
                published_item["collection"] = target_collection
                published_items.append(published_item)

            except Exception as e:
                # Re-raise with full item context for easier debugging
                raise RuntimeError(
                    f"Exception `{e}` while preparing/publishing "
                    f"item `{item.id}`: {json.dumps(item.to_dict(), indent=2)}",
                ) from e

    logger.info("End catalog publishing")
    return published_items

resolve_collection(item_metadata, generated_product_to_collection_identifier)

Resolve the target collection for a given product metadata.

The matching logic uses both output_product_id and product_type from the DprProcessedItemMetadata to find the corresponding FlowGeneratedProduct. Wildcards ('*') are supported for product_type in FlowGeneratedProduct.

Returns:

Type	Description
str	The resolved target collection name.

Raises:

Type	Description
ValueError	If the product cannot be resolved to any collection.

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

def resolve_collection(
    item_metadata: DprProcessedItemMetadata,
    generated_product_to_collection_identifier: list[FlowGeneratedProduct],
) -> str:
    """Resolve the target collection for a given product metadata.

    The matching logic uses both output_product_id and product_type from the
    DprProcessedItemMetadata to find the corresponding FlowGeneratedProduct.
    Wildcards ('*') are supported for product_type in FlowGeneratedProduct.

    Returns:
        The resolved target collection name.

    Raises:
        ValueError: If the product cannot be resolved to any collection.
    """
    logger = get_logger()
    logger.info(
        f"Resolving target collection for item metadata: {item_metadata}"
        f" with generated_product_to_collection_identifier: {generated_product_to_collection_identifier}",
    )

    # We prioritize matches in the following order:
    # 1. Exact Name, Exact Type
    # 2. Exact Name, Wildcard Type (least specific)
    # wildcards for name are NOT supported, they don't make sense in this context as tasktable don't support them

    target_collection = None
    for gen_prod in generated_product_to_collection_identifier:
        if gen_prod.name != item_metadata.output_product_id:
            continue
        # 1. Exact Name, Exact Type
        if item_metadata.product_type == gen_prod.product_type:
            target_collection = gen_prod.collection_name or gen_prod.product_type
            logger.info(f"Exact match (Name & Type) found: {gen_prod}, resolved collection: {target_collection}")
            break
        # 2. Exact Name, Wildcard Type
        if gen_prod.product_type == "*" and target_collection is None:
            target_collection = gen_prod.collection_name or gen_prod.product_type
            # This protection is also set in build_output_products from payload_generator.py,
            # but it's good to have it here as well for safety or a future change in logic.
            if target_collection == "*":
                raise RuntimeError(
                    f"The product type in generated_product_to_collection_identifier cannot be '*' "
                    f"if the collection name is not specified for product '{gen_prod.name}'",
                )
            logger.info(
                f"Match with Exact Name & Wildcard Type: {gen_prod}, resolved collection: {target_collection}",
            )

    if not target_collection:
        raise ValueError(
            f"No match for {item_metadata.output_product_id} in "
            f"{generated_product_to_collection_identifier}. "
            f"Could not find a collection to publish the stac_item from: {item_metadata}",
        )

    return target_collection