Skip to content

rs_workflows/on_demand/common/events.md

<< Back to index

Common events of the processing chains.

emit_product_generated_events(owner_id, products, workflow)

Emit a product.<product type>.generated event for each generated product.

The events are related to the workflow of the processing that generated the products (workflow role): the automations of the systematic processing start the next processing of the chain only from the events of a systematic workflow. Their payload contains the owner and the STAC item of the product, without its assets and links, because the Prefect event payloads are limited in size.

Source code in docs/rs-client-libraries/rs_workflows/on_demand/common/events.py 
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
def emit_product_generated_events(
    owner_id: str,
    products: list[dict[str, Any]],
    workflow: WorkflowType,
) -> None:
    """
    Emit a `product.<product type>.generated` event for each generated product.

    The events are related to the workflow of the processing that generated the products (`workflow` role): the
    automations of the systematic processing start the next processing of the chain only from the events of a
    `systematic` workflow. Their payload contains the owner and the STAC item of the product, without its assets and
    links, because the Prefect event payloads are limited in size.
    """
    logger = get_logger()
    flow_run_id = str(runtime.flow_run.id or "unknown")
    workflow_name = WorkflowType(workflow).value
    for product in products:
        product_type = product["properties"].get("product:type")
        if not product_type:
            logger.warning(f"⚠️ No event emitted for product '{product['id']}' without product type")
            continue
        event_name = product_generated_event_name(product_type)
        related = [
            {
                "prefect.resource.id": workflow_name,
                "prefect.resource.role": "workflow",
            },
            {"prefect.resource.id": "prefect.tag.processing", "prefect.resource.role": "tag"},
        ]
        if platform := product["properties"].get("platform"):
            related.append({"prefect.resource.id": f"prefect.tag.{platform}", "prefect.resource.role": "tag"})
        event = emit_event(
            event=event_name,
            resource={
                "prefect.resource.id": f"prefect.flow-run.{flow_run_id}",
                "prefect.resource.name": product_type,
            },
            related=related,
            payload={
                "owner_id": owner_id,
                "product": {
                    "id": product["id"],
                    "geometry": product["geometry"],
                    "collection": product["collection"],
                    "properties": product["properties"],
                },
            },
        )

        if event is None:
            logger.warning(f"⚠️ Event '{event_name}' was not emitted")
        else:
            logger.info(f"Event '{event_name}' (id='{event.id}') emitted.")

product_generated_event_name(product_type)

Return the name of the event emitted for a generated product of the given type.

Source code in docs/rs-client-libraries/rs_workflows/on_demand/common/events.py 
28
29
30
def product_generated_event_name(product_type: str) -> str:
    """Return the name of the event emitted for a generated product of the given type."""
    return PRODUCT_GENERATED_EVENT.format(product_type=product_type)