Skip to content

Common source code

The following source code is used in both packages rs_client and rs_workflows:

We can't pass a RsClient instance to the workflow because it causes (de-)serialization issues. So instead we pass the instance parameters, that will be used to recreate a new instance from the workflow.

TODO: also (de-)serialize the pystac_client.Client.open(...) parameters ?

Attributes:

Name Type Description
cls Class

RsClient child class type
rs_server_href str

RS-Server URL. In local mode, pass None.
rs_server_api_key str

API key for RS-Server authentication.
owner_id str

ID of the owner of the STAC catalog collections (no special characters allowoed).
station ECadipStation

Cadip station (if applicable)
Source code in docs/rs-client-libraries/rs_common/serialization.py 
27
28
29
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
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
86
class RsClientSerialization:  # pylint: disable=too-few-public-methods
    """
    We can't pass a RsClient instance to the workflow because it causes (de-)serialization issues.
    So instead we pass the instance parameters, that will be used to recreate a new instance
    from the workflow.

    TODO: also (de-)serialize the pystac_client.Client.open(...) parameters ?

    Attributes:
        cls (Class): RsClient child class type
        rs_server_href (str): RS-Server URL. In local mode, pass None.
        rs_server_api_key (str): API key for RS-Server authentication.
        owner_id (str): ID of the owner of the STAC catalog collections (no special characters allowoed).
        station (ECadipStation): Cadip station (if applicable)

    """

    def __init__(self, client: RsClient):
        """
        Serialize a RsClient instance.

        Args:
            client (RsClient): RsClient instance
        """

        # Save the parameters, except the logging (which should not be (de-)serialized neither.
        self.cls: Type[RsClient] = type(client)
        self.rs_server_href: str | None = client.rs_server_href
        self.rs_server_api_key: str | None = client.rs_server_api_key
        self.owner_id: str | None = client.owner_id

        self.station: ECadipStation | None = None
        if self.cls == CadipClient:
            self.station = cast(CadipClient, client).station

    def deserialize(self, logger=None) -> RsClient:
        """
        Recreate a new RsClient instance from the serialized parameters.

        Return:
            client (RsClient): RsClient instance
            logger (logging.Logger): Logging instance.
        """

        # Init parent class
        client = RsClient(self.rs_server_href, self.rs_server_api_key, self.owner_id, logger)

        # Return child instance
        if self.cls == AuxipClient:
            return client.get_auxip_client()
        if self.cls == CadipClient:
            return client.get_cadip_client(self.station)  # type: ignore
        if self.cls == StacClient:
            return client.get_stac_client()
        raise ValueError(f"Unknown RsClient type: {self.cls}")

    @property
    def get_flow_name(self):
        """Return a unique flow name for Prefect."""
        return f"{self.owner_id}_{datetime.now(timezone.utc).strftime(DATETIME_FORMAT)}"

get_flow_name property

Return a unique flow name for Prefect.

__init__(client)

Serialize a RsClient instance.

Parameters:

Name Type Description Default
client RsClient

RsClient instance

 required
Source code in docs/rs-client-libraries/rs_common/serialization.py 
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
def __init__(self, client: RsClient):
    """
    Serialize a RsClient instance.

    Args:
        client (RsClient): RsClient instance
    """

    # Save the parameters, except the logging (which should not be (de-)serialized neither.
    self.cls: Type[RsClient] = type(client)
    self.rs_server_href: str | None = client.rs_server_href
    self.rs_server_api_key: str | None = client.rs_server_api_key
    self.owner_id: str | None = client.owner_id

    self.station: ECadipStation | None = None
    if self.cls == CadipClient:
        self.station = cast(CadipClient, client).station

deserialize(logger=None)

Recreate a new RsClient instance from the serialized parameters.

Return

client (RsClient): RsClient instance logger (logging.Logger): Logging instance.

Source code in docs/rs-client-libraries/rs_common/serialization.py 
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
def deserialize(self, logger=None) -> RsClient:
    """
    Recreate a new RsClient instance from the serialized parameters.

    Return:
        client (RsClient): RsClient instance
        logger (logging.Logger): Logging instance.
    """

    # Init parent class
    client = RsClient(self.rs_server_href, self.rs_server_api_key, self.owner_id, logger)

    # Return child instance
    if self.cls == AuxipClient:
        return client.get_auxip_client()
    if self.cls == CadipClient:
        return client.get_cadip_client(self.station)  # type: ignore
    if self.cls == StacClient:
        return client.get_stac_client()
    raise ValueError(f"Unknown RsClient type: {self.cls}")