Skip to content

rs_server_common/data_retrieval/provider.md

<< Back to index

Provider mechanism.

CreateProviderFailed

Bases: Exception

Exception raised when an error occurred during the init of a provider.

Source code in docs/rs-server/services/common/rs_server_common/data_retrieval/provider.py 
53
54
class CreateProviderFailed(Exception):
    """Exception raised when an error occurred during the init of a provider."""

DownloadProductFailed

Bases: Exception

Exception raised when an error occurred during the download.

Source code in docs/rs-server/services/common/rs_server_common/data_retrieval/provider.py 
61
62
class DownloadProductFailed(Exception):
    """Exception raised when an error occurred during the download."""

Product dataclass

A product.

A product has an external identifier and a dictionary of metadata.

Source code in docs/rs-server/services/common/rs_server_common/data_retrieval/provider.py 
42
43
44
45
46
47
48
49
50
@dataclass
class Product:
    """A product.

    A product has an external identifier and a dictionary of metadata.
    """

    id_: str
    metadata: dict[str, str]

Provider

Bases: ABC

A product provider.

A provider gives a common interface to search for files from an external data source and download them locally.

Source code in docs/rs-server/services/common/rs_server_common/data_retrieval/provider.py 
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
class Provider(ABC):
    """A product provider.

    A provider gives a common interface to search for files from an external data source
    and download them locally.
    """

    def search(self, **kwargs) -> Any:
        """Search for products with the given time range.

        The search result is a dictionary of products found indexed by id.

        Returns:
            The files found indexed by file id. Specific to each provider.

        """
        time = kwargs.get("datetime", [None, None])
        between = TimeRange(start, stop) if (start := time[1]) and (stop := time[2]) else None
        if between:
            if between.duration() == timedelta(0):
                return []
            if between.duration() < timedelta(0):
                raise SearchProductFailed(f"Search timerange is inverted : ({between.start} -> {between.end})")
        return self._specific_search(**kwargs)

    @abstractmethod
    def _specific_search(self, **kwargs) -> Any:
        """Search for products with the given query parameters.

        Specific search for products after common verification.

        Returns:
            the files found indexed by file id.

        """

    @abstractmethod
    def download(self, product_id: str, to_file: Path) -> None:
        """Download the given product to the given local path.

        Args:
            product_id: id of the product to download
            to_file: path where the file should be downloaded

        Returns:
            None

        """

download(product_id, to_file) abstractmethod

Download the given product to the given local path.

Parameters:

Name Type Description Default
product_id str

id of the product to download

 required
to_file Path

path where the file should be downloaded

 required

Returns:

Type Description
None

None
Source code in docs/rs-server/services/common/rs_server_common/data_retrieval/provider.py 
101
102
103
104
105
106
107
108
109
110
111
112
@abstractmethod
def download(self, product_id: str, to_file: Path) -> None:
    """Download the given product to the given local path.

    Args:
        product_id: id of the product to download
        to_file: path where the file should be downloaded

    Returns:
        None

    """

search(**kwargs)

Search for products with the given time range.

The search result is a dictionary of products found indexed by id.

Returns:

Type Description
Any

The files found indexed by file id. Specific to each provider.
Source code in docs/rs-server/services/common/rs_server_common/data_retrieval/provider.py 
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
def search(self, **kwargs) -> Any:
    """Search for products with the given time range.

    The search result is a dictionary of products found indexed by id.

    Returns:
        The files found indexed by file id. Specific to each provider.

    """
    time = kwargs.get("datetime", [None, None])
    between = TimeRange(start, stop) if (start := time[1]) and (stop := time[2]) else None
    if between:
        if between.duration() == timedelta(0):
            return []
        if between.duration() < timedelta(0):
            raise SearchProductFailed(f"Search timerange is inverted : ({between.start} -> {between.end})")
    return self._specific_search(**kwargs)

SearchProductFailed

Bases: Exception

Exception raised when an error occurred during the search.

Source code in docs/rs-server/services/common/rs_server_common/data_retrieval/provider.py 
57
58
class SearchProductFailed(Exception):
    """Exception raised when an error occurred during the search."""

TimeRange dataclass

A time range.

Source code in docs/rs-server/services/common/rs_server_common/data_retrieval/provider.py 
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
@dataclass
class TimeRange:
    """A time range."""

    start: datetime
    end: datetime

    def duration(self) -> timedelta:
        """Duration of the timerange.

        Returns: duration of the timerange
        """
        return self.end - self.start

    def __bool__(self) -> bool:
        return self.start is not None and self.end is not None

duration()

Duration of the timerange.

Returns: duration of the timerange

Source code in docs/rs-server/services/common/rs_server_common/data_retrieval/provider.py 
31
32
33
34
35
36
def duration(self) -> timedelta:
    """Duration of the timerange.

    Returns: duration of the timerange
    """
    return self.end - self.start