rs_workflows/earthdatahub_flow.md

<< Back to index

EarthDataHub flow implementation.

earthdatahub_search_task(*args, **kwargs) async

See: search

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

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

search(env, edh_cql2, error_if_empty=False) async

Search STAC products in EarthDataHub catalogue.

Parameters:

Name	Type	Description	Default
env	FlowEnvArgs	Prefect flow environment (at least the owner_id is required)	required
edh_cql2	dict	CQL2 filter read from the processor tasktable.	required
error_if_empty	bool	Raise a ValueError if the results are empty.	False

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

@flow(name="search-earthdatahub")
async def search(
    env: FlowEnvArgs,
    edh_cql2: dict,
    error_if_empty: bool = False,
) -> ItemCollection | None:
    """
    Search STAC products in EarthDataHub catalogue.

    Args:
        env: Prefect flow environment (at least the owner_id is required)
        edh_cql2: CQL2 filter read from the processor tasktable.
        error_if_empty: Raise a ValueError if the results are empty.
    """
    return await stac.search(
        env=env,
        cql2=edh_cql2,
        span_name="earthdatahub-search",
        stac_client_selector=lambda flow_env: (flow_env.rs_client.get_earthdatahub_client(), {}),
        error_if_empty=error_if_empty,
    )