rs_workflows/flow_utils.md

<< Back to index

Utility module for the Prefect flows.

AdfProcessIn

Bases: BaseModel

Input parameters for executing the 'adf_conversion' flow.

This model defines all the configuration needed to run an ADF conversion script.

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

class AdfProcessIn(BaseModel):
    """
    Input parameters for executing the 'adf_conversion' flow.

    This model defines all the configuration needed to run an ADF conversion script.
    """

    env: FlowEnvArgs = Field(
        title="Flow Environment",
        description="Environment configuration for Prefect flow. Includes identifiers like owner_id.",
    )
    adf_type: str | AdfType = Field(
        title="ADF Type",
        description="Name of the ADF type to generate. Can be a string or AdfType enum.",
    )
    auxiliary_product_to_collection_identifier: list[AuxiliaryProductMapping] = Field(
        title="Auxiliary Product Mapping",
        description=(
            "Mapping of auxiliary product types to collections. "
            "Use '*' as a wildcard to map all other auxiliary products."
        ),
        min_length=1,
    )
    start_datetime: datetime | None = Field(
        default=None,
        title="Start Datetime",
        description="Start datetime for retrieving auxiliary data. ISO format.",
    )
    end_datetime: datetime | None = Field(
        default=None,
        title="End Datetime",
        description="End datetime for retrieving auxiliary data. ISO format.",
    )
    satellite: str | SentinelSatellite | None = Field(
        default=None,
        title="Satellite",
        description="Satellite identifier used in certain queries. Can be a string or SentinelSatellite enum. "
        "Mandatory for the mission-dependent ADF types (orbit files).",
    )
    cql2_filter: dict | None = Field(
        default=None,
        title="CQL2 Filter",
        description="CQL2 filter for retrieving auxiliary data."
        "If provided, start_datetime, end_datetime, and satellite will be ignored for auxiliary data retrieval.",
    )

    @field_validator("adf_type", mode="before")
    @classmethod
    def normalize_adf_type(cls, v):
        """Normalize ADF type to string."""
        return v.value if hasattr(v, "value") else v

    @field_validator("satellite", mode="before")
    @classmethod
    def normalize_satellite_name(cls, v):
        """Normalize satellite name to string."""
        return v.value if hasattr(v, "value") else v

normalize_adf_type(v) classmethod

Normalize ADF type to string.

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

@field_validator("adf_type", mode="before")
@classmethod
def normalize_adf_type(cls, v):
    """Normalize ADF type to string."""
    return v.value if hasattr(v, "value") else v

normalize_satellite_name(v) classmethod

Normalize satellite name to string.

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

@field_validator("satellite", mode="before")
@classmethod
def normalize_satellite_name(cls, v):
    """Normalize satellite name to string."""
    return v.value if hasattr(v, "value") else v

AdfType

Bases: str, Enum

ADF type

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

class AdfType(str, Enum):
    """ADF type"""

    S00__ADF_ECMWA = "S00__ADF_ECMWA"
    S00__ADF_ECMWF = "S00__ADF_ECMWF"
    S00__ADF_FPOAX = "S00__ADF_FPOAX"
    S00__ADF_FROAX = "S00__ADF_FROAX"
    S00__ADF_GETAS = "S00__ADF_GETAS"
    S00__ADF_OSFAX = "S00__ADF_OSFAX"
    S00__ADF_WATER = "S00__ADF_WATER"
    S03_ADF_OLCAL = "S03_ADF_OLCAL"
    S03_ADF_OLEOP = "S03_ADF_OLEOP"
    S03_ADF_OLINS = "S03_ADF_OLINS"
    S03_ADF_OLLUT = "S03_ADF_OLLUT"
    S03_ADF_OLPRG = "S03_ADF_OLPRG"
    S03_ADF_OLRAC = "S03_ADF_OLRAC"
    S03_ADF_OLSPC = "S03_ADF_OLSPC"
    S03_ADF_OLPPP = "S03_ADF_OLPPP"
    S03_ADF_OLPCP = "S03_ADF_OLPCP"
    S03_ADF_OLCLP = "S03_ADF_OLCLP"
    S03_ADF_OLWVP = "S03_ADF_OLWVP"
    S03_ADF_OLVGP = "S03_ADF_OLVGP"
    S03_ADF_SL1PP = "S03_ADF_SL1PP"
    S03_ADF_SLADJ = "S03_ADF_SLADJ"
    S03_ADF_SLANC = "S03_ADF_SLANC"
    S03_ADF_SLCDP = "S03_ADF_SLCDP"
    S03_ADF_SLCLO = "S03_ADF_SLCLO"
    S03_ADF_SLCLP = "S03_ADF_SLCLP"
    S03_ADF_SLGEC = "S03_ADF_SLGEC"
    S03_ADF_SLGEO = "S03_ADF_SLGEO"
    S03_ADF_SLVIC = "S03_ADF_SLVIC"
    S03_ADF_TIRCD = "S03_ADF_TIRCD"
    S03_ADF_VSWCD = "S03_ADF_VSWCD"

AuxiliaryProductMapping

Bases: BaseModel

Represents mapping for auxiliary products.

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

class AuxiliaryProductMapping(BaseModel):
    """Represents mapping for auxiliary products."""

    product_type: str = Field(description="Product type or '*' wildcard.")
    collection_name: str = Field(description="Collection name.")
    source: AuxiliarySource = Field(
        default=AuxiliarySource.AUXIP,
        description="STAC source where auxiliary products are searched.",
    )
    selected_assets: list[str] | None = Field(
        default=None,
        description="Optional asset keys to stage.",
    )

    def items(self):
        """Helper method to return the model fields as items, useful for logging."""
        return self.model_dump().items()

items()

Helper method to return the model fields as items, useful for logging.

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

def items(self):
    """Helper method to return the model fields as items, useful for logging."""
    return self.model_dump().items()

AuxiliarySource

Bases: str, Enum

STAC source for auxiliary product search.

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

class AuxiliarySource(str, Enum):
    """STAC source for auxiliary product search."""

    LTA = "lta"
    PRIP = "prip"
    CADIP = "cadip"
    AUXIP = "auxip"
    CDSE = "cdse"
    CATALOG = "catalog"
    EARTHDATAHUB = "earthdatahub"

ConversionIn

Bases: BaseModel

Input parameters for the 'convert-safe' Prefect flow, implemented by 'on_demand_conversion'.

Defines the configuration needed to convert a legacy SAFE product to its EOPF (Zarr) counterpart on demand, including the input product, output collection, Dask cluster, and output Zarr format.

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

class ConversionIn(BaseModel):
    """
    Input parameters for the 'convert-safe' Prefect flow, implemented by 'on_demand_conversion'.

    Defines the configuration needed to convert a legacy SAFE product to its EOPF (Zarr)
    counterpart on demand, including the input product, output collection, Dask cluster,
    and output Zarr format.
    """

    env: FlowEnvArgs = Field(
        title="Flow Environment",
        description="Environment configuration for Prefect flow. Includes identifiers like owner_id.",
    )
    stac_input: str | dict = Field(
        title="STAC Input Product",
        description=("Input product for the conversion. Specifies the STAC item or href."),
    )
    generated_product_to_collection_identifier: FlowGeneratedProduct = Field(
        title="Generated Product",
        description=(
            "Generated product. Specifies a name, the product type, and the collection where the output will be stored."
        ),
    )
    dask_cluster_label: str = Field(
        title="Dask Cluster Label",
        description="Label of the Dask cluster to use for SAFE conversion.",
    )
    dask_cluster_instance: str | None = Field(
        default=None,
        title="Dask Cluster Instance",
        description="Optional Dask cluster instance ID used by the DPR conversion service.",
    )
    selected_assets: list[str] | None = Field(
        default=None,
        title="Selected Assets",
        description=("Set of selected asset keys to stage. If not provided, all assets will be converted"),
    )
    zarr_format: Literal[2, 3] = Field(
        default=2,
        title="Zarr format (default: 2)",
        description="Output Zarr format. Defaults to 2 when omitted. ",
    )

DprProcessIn

Bases: BaseModel

Input parameters for executing the 'dpr-process' flow.

This model defines all the configuration needed to run a DPR processor, including input datasets, generated outputs, auxiliary data mapping, processing modes, and scheduling parameters.

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

class DprProcessIn(BaseModel):
    """
    Input parameters for executing the 'dpr-process' flow.

    This model defines all the configuration needed to run a DPR processor,
    including input datasets, generated outputs, auxiliary data mapping,
    processing modes, and scheduling parameters.
    """

    env: FlowEnvArgs = Field(
        title="Flow Environment",
        description="Environment configuration for Prefect flow. Includes identifiers like owner_id.",
    )

    processor_name: str | DprProcessor = Field(
        title="DPR Processor Name",
        description="Name of the DPR processor to run. Can be a string or DprProcessor enum.",
    )

    processor_version: str = Field(
        title="Processor Version",
        description="Version of the processor. If not relevant, can be empty string.",
    )

    dask_cluster_label: str = Field(
        title="Dask Cluster Label",
        description='Label of the Dask cluster to use, e.g. "dask-l0" for local testing.',
    )

    dask_cluster_instance: str | None = Field(
        default=None,
        title="Dask Cluster Instance",
        description="Optional Dask cluster instance ID used to build a direct dashboard URL.",
    )

    dask_task_timeout: int | None = Field(
        default=None,
        title="Dask task timeout",
        description="Default timeout on a submitted task",
    )

    temporary_shared: bool = Field(
        default=False,
        title="Temporary folder shared",
        description="Whether the temporary folder is reachable from the workers",
    )

    temporary_folder: str | None = Field(
        default=None,
        title="Temporary folder",
        description="Temporary folder path",
    )

    triggering__stage_s3_outputs: bool = Field(
        default=True,
        title="Stage S3 Outputs",
        description=(
            "Experimental CPM feature: save S3 workflow outputs to a local temporary product, "
            "then upload it to S3. Disable to use direct S3 writes."
        ),
    )

    triggering__stage_s3_temporary_prefix: bool = Field(
        default=False,
        title="Stage S3 Temporary Prefix",
        description=(
            "Experimental CPM feature: when S3 output staging is enabled, upload to a temporary S3 prefix, "
            "copy to the final prefix, then remove the temporary prefix. Adds an S3 copy; promotion is not atomic."
        ),
    )

    triggering__stage_s3_memory_fit_safety_ratio: float = Field(
        default=0.25,
        title="Stage S3 Memory Fit Safety Ratio",
        description=(
            "Fraction of available runner (client) memory used as the threshold for staged S3 Zarr outputs "
            "with a remote Dask scheduler. Products estimated to exceed this threshold use the experimental "
            "Sub-DataTree writer. Lower values select this mode for smaller products. "
            "The threshold uses runner memory, not scheduler memory."
        ),
    )

    triggering__stage_s3_outputs_min_size: int = Field(
        default=0,
        ge=0,
        title="Stage S3 Outputs Minimum Size",
        description=(
            "Minimum logical product size in bytes for experimental CPM S3 output staging. "
            "0 stages every S3 output; smaller products use direct writes."
        ),
    )

    s3_payload_file: str = Field(
        title="S3 Payload File",
        description="S3 or local path where the processor payload will be written for execution.",
    )

    pipeline: str | DprPipeline | None = Field(
        default=None,
        title="Pipeline Name",
        description="Name of the processing pipeline. Must be provided if `unit` is not set.",
    )

    unit: str | None = Field(
        default=None,
        title="Unit Name",
        description="Processing unit name. Must be provided if `pipeline` is not set.",
    )

    priority: Priority = Field(
        default=Priority.LOW,
        title="Processing Priority",
        description="Priority to assign for processing on the Dask cluster.",
    )

    workflow_type: WorkflowType = Field(
        default=WorkflowType.ON_DEMAND,
        title="Workflow Type",
        description="Type of workflow: ON_DEMAND, BENCHMARKING, SYSTEMATIC.",
    )

    input_products: list[FlowInputProduct] = Field(
        title="Input Products",
        description=(
            "List of input products for the processor. Each item specifies the product name, "
            "the STAC item identifier, and the collection it belongs to."
        ),
        min_length=1,
    )

    generated_product_to_collection_identifier: list[FlowGeneratedProduct] = Field(
        title="Generated Products",
        description=(
            "List of generated products. Each item specifies a name, the product type, "
            "and the collection where the output will be stored."
        ),
        min_length=1,
    )

    auxiliary_product_to_collection_identifier: list[AuxiliaryProductMapping] = Field(
        title="Auxiliary Product Mapping",
        description=(
            "Mapping of auxiliary product types to collections. "
            "Use '*' as a wildcard to map all other auxiliary products."
        ),
        min_length=1,
    )

    processing_mode: list[ProcessingMode] = Field(
        default_factory=list,
        title="Processing Modes",
        description="List of processing modes that control DPR behavior, e.g., ALWAYS, CONDITIONAL.",
    )

    start_datetime: datetime | None = Field(
        default=None,
        title="Start Datetime",
        description="Start datetime for retrieving auxiliary data. ISO format.",
    )

    end_datetime: datetime | None = Field(
        default=None,
        title="End Datetime",
        description="End datetime for retrieving auxiliary data. ISO format.",
    )

    satellite: str | SentinelSatellite | None = Field(
        default=None,
        title="Satellite",
        description="Satellite identifier used in certain queries. Can be a string or SentinelSatellite enum.",
    )

    reference_date: date | None = Field(
        default=None,
        title="Reference Date",
        description="Date used to identify a specific reference/master input product within the list of inputs.",
    )

    instrument_mode: str | InstrumentMode | None = Field(
        default=None,
        title="Instrument Mode",
        description="Instrument mode used in certain queries. Can be a string or InstrumentMode enum.",
    )

    edh_api_key: str | None = Field(
        default=None,
        title="EarthDataHub Standard API key",
        description="Destination Earth / EarthDataHub standard API key used to access Copernicus DEM",
    )

    # -----------------------
    # Validators
    # -----------------------

    @field_validator("processor_name", mode="before")
    @classmethod
    def normalize_processor_name(cls, v):
        """Normalize processor name to string."""
        return v.value if hasattr(v, "value") else v

    @field_validator("satellite", mode="before")
    @classmethod
    def normalize_satellite_name(cls, v):
        """Normalize satellite name to string."""
        return v.value if hasattr(v, "value") else v

    @model_validator(mode="after")
    def check_model(self):
        """Ensure mutual exclusivity between pipeline and unit."""
        has_pipeline = bool(self.pipeline)
        has_unit = bool(self.unit)

        if has_pipeline == has_unit:
            raise ValueError("Exactly one of 'pipeline' or 'unit' must be provided.")

        return self

check_model()

Ensure mutual exclusivity between pipeline and unit.

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

@model_validator(mode="after")
def check_model(self):
    """Ensure mutual exclusivity between pipeline and unit."""
    has_pipeline = bool(self.pipeline)
    has_unit = bool(self.unit)

    if has_pipeline == has_unit:
        raise ValueError("Exactly one of 'pipeline' or 'unit' must be provided.")

    return self

normalize_processor_name(v) classmethod

Normalize processor name to string.

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

@field_validator("processor_name", mode="before")
@classmethod
def normalize_processor_name(cls, v):
    """Normalize processor name to string."""
    return v.value if hasattr(v, "value") else v

normalize_satellite_name(v) classmethod

Normalize satellite name to string.

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

@field_validator("satellite", mode="before")
@classmethod
def normalize_satellite_name(cls, v):
    """Normalize satellite name to string."""
    return v.value if hasattr(v, "value") else v

DprProcessOut dataclass

Output parameters for the 'dpr-process' flow

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

@dataclass
class DprProcessOut:
    """
    Output parameters for the 'dpr-process' flow
    """

    status: bool
    product_identifier: list[Item] = field(default_factory=list)

DprProcessedItemMetadata dataclass

Metadata for a DPR processed item.

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

@dataclass
class DprProcessedItemMetadata:
    """Metadata for a DPR processed item."""

    output_product_id: str
    product_type: str | None
    stac_item: Item

FlowEnv

Prefect flow environment and reusable objects.

Attributes:

Name	Type	Description
owner_id	str	User/owner ID
calling_span	SpanContext | None	OpenTelemetry span of the calling flow, if any.
this_span	SpanContext | None	Current OpenTelemetry span.
rs_client	RsClient	RsClient instance

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

class FlowEnv:
    """
    Prefect flow environment and reusable objects.

    Attributes:
        owner_id (str): User/owner ID
        calling_span (SpanContext | None): OpenTelemetry span of the calling flow, if any.
        this_span (SpanContext | None): Current OpenTelemetry span.
        rs_client (RsClient): RsClient instance
    """

    def __init__(self, args: FlowEnvArgs):
        """Constructor."""
        self.owner_id: str = args.owner_id
        self.calling_span: SpanContext | None = None
        self.this_span: SpanContext | None = None
        self.logging_level: str = args.logging_level.value

        # Set root logging level to the one selected for this flow
        logging.getLogger().setLevel(self.logging_level)

        # Set logging level for flows and tasks
        logging.getLogger("prefect.flow_runs").setLevel(self.logging_level)
        logging.getLogger("prefect.task_runs").setLevel(self.logging_level)

        logger = get_logger()
        logger.info("Initializing FlowEnv with args: %r", args)

        # Deserialize the calling span, if any
        if args.calling_span:
            self.calling_span = SpanContext(*args.calling_span)

        # Read prefect blocks into env vars
        prefect_utils.read_prefect_blocks(self.owner_id, _sync=True)  # type: ignore
        logger.info("Prefect blocks read into env vars for owner_id=%r", self.owner_id)
        # Init opentelemetry traces
        init_opentelemetry.init_traces(args.service_name)

        # Init the RsClient instance from the env vars
        self.rs_client = RsClient(
            rs_server_href=os.getenv("RSPY_WEBSITE"),
            rs_server_api_key=os.getenv("RSPY_APIKEY"),
            owner_id=self.owner_id,
            logger=get_logger(),  # type: ignore
        )

    def serialize(self) -> FlowEnvArgs:
        """Serialize this object with Pydantic."""

        # The serialized object will be used by a new opentelemetry span.
        # Its calling span will be either the current span, or the current calling span.
        new_calling_span = self.this_span or self.calling_span
        if new_calling_span:
            # Only keep the first n attributes, the other need custom serialization
            serialized_span = tuple(new_calling_span)[:3]
        else:
            serialized_span = None
        logging_level = LoggingLevel(self.logging_level) or LoggingLevel.INFO

        return FlowEnvArgs(
            owner_id=self.owner_id,
            calling_span=serialized_span,
            logging_level=logging_level,
        )  # type: ignore

    @_agnosticcontextmanager
    def start_span(
        self,
        instrumenting_module_name: str,
        name: str,
    ) -> Iterator[Span]:
        """
        Context manager for creating a new main or child OpenTelemetry span and set it
        as the current span in this tracer's context.

        Args:
            instrumenting_module_name: Caller module name, just pass __name__
            name: The name of the span to be created (use a custom name)

        Yields:
            The newly-created span.
        """
        # Create new span and save it
        with init_opentelemetry.start_span(  # pylint: disable=contextmanager-generator-missing-cleanup
            instrumenting_module_name,
            name,
            self.calling_span,
        ) as span:
            self.this_span = trace.get_current_span().get_span_context()
            yield span

__init__(args)

Constructor.

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

def __init__(self, args: FlowEnvArgs):
    """Constructor."""
    self.owner_id: str = args.owner_id
    self.calling_span: SpanContext | None = None
    self.this_span: SpanContext | None = None
    self.logging_level: str = args.logging_level.value

    # Set root logging level to the one selected for this flow
    logging.getLogger().setLevel(self.logging_level)

    # Set logging level for flows and tasks
    logging.getLogger("prefect.flow_runs").setLevel(self.logging_level)
    logging.getLogger("prefect.task_runs").setLevel(self.logging_level)

    logger = get_logger()
    logger.info("Initializing FlowEnv with args: %r", args)

    # Deserialize the calling span, if any
    if args.calling_span:
        self.calling_span = SpanContext(*args.calling_span)

    # Read prefect blocks into env vars
    prefect_utils.read_prefect_blocks(self.owner_id, _sync=True)  # type: ignore
    logger.info("Prefect blocks read into env vars for owner_id=%r", self.owner_id)
    # Init opentelemetry traces
    init_opentelemetry.init_traces(args.service_name)

    # Init the RsClient instance from the env vars
    self.rs_client = RsClient(
        rs_server_href=os.getenv("RSPY_WEBSITE"),
        rs_server_api_key=os.getenv("RSPY_APIKEY"),
        owner_id=self.owner_id,
        logger=get_logger(),  # type: ignore
    )

serialize()

Serialize this object with Pydantic.

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

def serialize(self) -> FlowEnvArgs:
    """Serialize this object with Pydantic."""

    # The serialized object will be used by a new opentelemetry span.
    # Its calling span will be either the current span, or the current calling span.
    new_calling_span = self.this_span or self.calling_span
    if new_calling_span:
        # Only keep the first n attributes, the other need custom serialization
        serialized_span = tuple(new_calling_span)[:3]
    else:
        serialized_span = None
    logging_level = LoggingLevel(self.logging_level) or LoggingLevel.INFO

    return FlowEnvArgs(
        owner_id=self.owner_id,
        calling_span=serialized_span,
        logging_level=logging_level,
    )  # type: ignore

start_span(instrumenting_module_name, name)

Context manager for creating a new main or child OpenTelemetry span and set it as the current span in this tracer's context.

Parameters:

Name	Type	Description	Default
instrumenting_module_name	str	Caller module name, just pass name	required
name	str	The name of the span to be created (use a custom name)	required

Yields:

Type	Description
Span	The newly-created span.

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

@_agnosticcontextmanager
def start_span(
    self,
    instrumenting_module_name: str,
    name: str,
) -> Iterator[Span]:
    """
    Context manager for creating a new main or child OpenTelemetry span and set it
    as the current span in this tracer's context.

    Args:
        instrumenting_module_name: Caller module name, just pass __name__
        name: The name of the span to be created (use a custom name)

    Yields:
        The newly-created span.
    """
    # Create new span and save it
    with init_opentelemetry.start_span(  # pylint: disable=contextmanager-generator-missing-cleanup
        instrumenting_module_name,
        name,
        self.calling_span,
    ) as span:
        self.this_span = trace.get_current_span().get_span_context()
        yield span

FlowEnvArgs

Bases: BaseModel

Prefect flow environment arguments.

Attributes:

Name	Type	Description
owner_id	str	User/owner ID (necessary to retrieve the user info: API key and OAuth2 cookie)
from	the right Prefect block. NOTE	may be useless after each user has their own prefect
calling_span	tuple	Serialized OpenTelemetry span of the calling flow, if any.
service_name	str	OpenTelemetry service name

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

class FlowEnvArgs(BaseModel):
    """
    Prefect flow environment arguments.

    Attributes:
        owner_id: User/owner ID (necessary to retrieve the user info: API key and OAuth2 cookie)
        from the right Prefect block. NOTE: may be useless after each user has their own prefect
        server because there will be only one block.
        calling_span (tuple): Serialized OpenTelemetry span of the calling flow, if any.
        service_name: OpenTelemetry service name
    """

    owner_id: str = Field(
        description="User/owner ID (necessary to retrieve the user info) from the right Prefect block",
    )
    logging_level: LoggingLevel = Field(
        default=(
            LoggingLevel[DEFAULT_ROOT_LOGGING_LEVEL]
            if DEFAULT_ROOT_LOGGING_LEVEL in LoggingLevel
            else LoggingLevel.INFO
        ),
        title="Overall logging level",
        description="Level of logs wanted for this flow. Also applied to EOPF logs.",
    )
    calling_span: tuple[int, int, bool] | None = Field(
        default=None,
        description="Serialized OpenTelemetry span of the calling flow, if any",
    )
    service_name: str = Field(default="rs.workflows", description="OpenTelemetry service name")
    called_by: str = Field(
        default="",
        title="Called by",
        description="Any additional information on who or what called the flow",
    )

FlowGeneratedProduct

Bases: BaseModel

Represents one generated output product.

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

class FlowGeneratedProduct(BaseModel):
    """Represents one generated output product."""

    name: str = Field(description="Output product name.")
    product_type: str = Field(description="Product type.")
    collection_name: str | None = Field(
        default=None,
        description="Collection name. If not provided, it defaults to product_type.",
    )

    def items(self):
        """Helper method to return the model fields as items, useful for logging."""
        return self.model_dump().items()

items()

Helper method to return the model fields as items, useful for logging.

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

def items(self):
    """Helper method to return the model fields as items, useful for logging."""
    return self.model_dump().items()

FlowInputProduct

Bases: BaseModel

Represents one input product for the processor.

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

class FlowInputProduct(BaseModel):
    """Represents one input product for the processor."""

    name: str = Field(description="Input product name.")
    item_id: str = Field(description="STAC item identifier.")
    collection_name: str = Field(description="Collection name.")

    def items(self):
        """Helper method to return the model fields as items, useful for logging."""
        return self.model_dump().items()

items()

Helper method to return the model fields as items, useful for logging.

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

def items(self):
    """Helper method to return the model fields as items, useful for logging."""
    return self.model_dump().items()

InstrumentMode

Bases: str, Enum

Instrument mode

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

class InstrumentMode(str, Enum):
    """Instrument mode"""

    EW = "EW"
    IW = "IW"
    SM = "SM"

LoggingLevel

Bases: str, Enum

Logging level allowed by eopf.logging module

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

class LoggingLevel(str, Enum):
    """Logging level allowed by eopf.logging module"""

    DEBUG = "DEBUG"
    INFO = "INFO"
    WARNING = "WARNING"
    ERROR = "ERROR"
    CRITICAL = "CRITICAL"

Priority

Bases: str, Enum

Priority for the cluster dask to be able to prioritise task execution.

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

class Priority(str, Enum):
    """
    Priority for the cluster dask to be able to prioritise task execution.
    """

    LOW = "low"
    MEDIUM = "medium"
    HIGH = "high"

ProcessingMode

Bases: str, Enum

List of mode to be applied when calling the DPR processor.

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

class ProcessingMode(str, Enum):
    """
    List of mode to be applied when calling the DPR processor.
    """

    NRT = "nrt"
    NTC = "ntc"
    REPROCESSING = "reprocessing"
    SUBS = "subs"
    ALWAYS = "always"

RetryConfig

Bases: BaseModel

Parameters:

Name	Type	Description	Default
staging_retries		Number of retry attempts for staging operations	required
staging_retry_delay		Delay in seconds between retry attempts.	required

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

class RetryConfig(BaseModel):
    """
    Args:
        staging_retries: Number of retry attempts for staging operations
        staging_retry_delay: Delay in seconds between retry attempts.
    """

    staging_retries: int = Field(3, description="Number of retry attempts for staging operations.")
    staging_retry_delay: int = Field(60, description="Delay in seconds between retry attempts.")

SentinelSatellite

Bases: str, Enum

Sentinel satellite name

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

class SentinelSatellite(str, Enum):
    """Sentinel satellite name"""

    # String value = STAC standardized value
    S1A = "sentinel-1a"
    S1B = "sentinel-1b"
    S1C = "sentinel-1c"
    S1D = "sentinel-1d"
    S2A = "sentinel-2a"
    S2B = "sentinel-2b"
    S2C = "sentinel-2c"
    S3A = "sentinel-3a"
    S3B = "sentinel-3b"

WorkflowType

Bases: str, Enum

Workflow type.

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

class WorkflowType(str, Enum):
    """
    Workflow type.
    """

    BENCHMARKING = "benchmarking"
    ON_DEMAND = "on-demand"
    SYSTEMATIC = "systematic"

    @classmethod
    def of_run(cls, force_systematic: bool) -> "WorkflowType":
        """
        Workflow of a run started by a user (on-demand), or by an automation of the systematic processing
        (`force_systematic`): the automations start the next processing of the chain only from a systematic one.
        """
        return cls.SYSTEMATIC if force_systematic else cls.ON_DEMAND

of_run(force_systematic) classmethod

Workflow of a run started by a user (on-demand), or by an automation of the systematic processing (force_systematic): the automations start the next processing of the chain only from a systematic one.

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

@classmethod
def of_run(cls, force_systematic: bool) -> "WorkflowType":
    """
    Workflow of a run started by a user (on-demand), or by an automation of the systematic processing
    (`force_systematic`): the automations start the next processing of the chain only from a systematic one.
    """
    return cls.SYSTEMATIC if force_systematic else cls.ON_DEMAND