Skip to content

gportal.search

search(dataset_ids=[], bbox=None, start_time=None, end_time=None, count=100, params={}, timeout=120)

Searches products on G-Portal with the given parameters.

Note that the return value is a Search instance. See its documentation for details on how to get the number of matches and iterate over results.

Parameters:

Name Type Description Default
dataset_ids Iterable[Union[str, int]]

List of dataset IDs.

 []
bbox Optional[Sequence[float]]

Bounding box of coordinates.

 None
start_time Union[str, datetime, None]

Observation start time.

 None
end_time Union[str, datetime, None]

Observation end time.

 None
count int

Number of products per page.

 100
params dict[str, Any]

Additional search parameters. See G-Portal User's Manual.

 {}
timeout Optional[float]

Timeout in seconds.

 120

Returns:

Type Description
Search

A Search instance that represents deferred query.
Source code in gportal/search.py 
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
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
def search(
    dataset_ids: Iterable[Union[str, int]] = [],
    bbox: Optional[Sequence[float]] = None,
    start_time: Union[str, datetime, None] = None,
    end_time: Union[str, datetime, None] = None,
    count: int = 100,
    params: dict[str, Any] = {},
    timeout: Optional[float] = 120,
) -> "Search":
    """Searches products on G-Portal with the given parameters.

    Note that the return value is a [`Search`][gportal.search.Search] instance.
    See its documentation for details on how to get the number of matches and iterate over results.

    Args:
        dataset_ids: List of dataset IDs.
        bbox: Bounding box of coordinates.
        start_time: Observation start time.
        end_time: Observation end time.
        count: Number of products per page.
        params: Additional search parameters. See G-Portal User's Manual.
        timeout: Timeout in seconds.

    Returns:
        A [Search][gportal.search.Search] instance that represents deferred query.
    """
    params = params.copy()

    if dataset_ids:
        params["datasetId"] = ",".join(map(str, dataset_ids))

    if bbox is not None:
        params["bbox"] = ",".join(map(str, bbox))

    if isinstance(start_time, datetime):
        params["startTime"] = start_time.isoformat()
    elif start_time:
        params["startTime"] = start_time

    if isinstance(end_time, datetime):
        params["endTime"] = end_time.isoformat()
    elif end_time:
        params["endTime"] = end_time

    params["count"] = count

    return Search(params, timeout=timeout)

Search

Search query for G-Portal Catalogue Service API.

For API details, see G-Portal User's Manual Appendix 7: https://gportal.jaxa.jp/gpr/assets/mng_upload/COMMON/upload/GPortalUserManual_en.pdf

Attributes:

Name Type Description
params Mapping[str, Any]

Search parameters.
timeout Optional[float]

Timeout in seconds.
Source code in gportal/search.py 
 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
 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
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
class Search:
    """Search query for G-Portal Catalogue Service API.

    For API details, see G-Portal User's Manual Appendix 7:
    https://gportal.jaxa.jp/gpr/assets/mng_upload/COMMON/upload/GPortalUserManual_en.pdf

    Attributes:
        params: Search parameters.
        timeout: Timeout in seconds.
    """

    def __init__(self, params: Mapping[str, Any], timeout: Optional[float] = None):
        self.params: Mapping[str, Any] = params
        self.timeout: Optional[float] = timeout

    def __repr__(self) -> str:
        return f"<gportal.Search params={self.params}>"

    @cache
    def matched(self) -> Optional[int]:
        """Gets the number of products matched the query.

        This method is cached, so it is safe to call it multiple times.

        Returns:
            The number of products matched the query.
        """
        response = self._request({"count": 0})
        return response.get("properties", {}).get("numberOfRecordsMatched")

    def products(self, convert_types: bool = True) -> Iterator[Product]:
        """Yields products matched the query.

        Args:
            convert_types: Whether to convert string values in properties into int, float, bool, or None.

        Yields:
            A [Product][gportal.product.Product] instance.
        """
        for page in self.pages():
            for product_dict in page.get("features", []):
                yield Product(product_dict, convert_types=convert_types)

    def pages(self) -> Iterator[dict[str, Any]]:
        """Iterates a search request with pagination.

        Yields:
            A dictionary of GeoJSON FeatureCollection.
        """
        start_index = 1

        while True:
            page = self._request({"startIndex": start_index})
            yield page

            properties = page["properties"]
            start_index += properties["numberOfRecordsReturned"]

            if start_index > properties["numberOfRecordsMatched"]:
                break

    def _request(self, extra_params: dict[str, Any]) -> dict[str, Any]:
        return http_client.get(
            "/csw/csw",
            params={
                **self.params,
                "service": "CSW",
                "version": "3.0.0",
                "request": "GetRecords",
                "outputFormat": "application/json",
                **extra_params,
            },
            timeout=self.timeout,
        )

matched() cached

Gets the number of products matched the query.

This method is cached, so it is safe to call it multiple times.

Returns:

Type Description
Optional[int]

The number of products matched the query.
Source code in gportal/search.py 
77
78
79
80
81
82
83
84
85
86
87
@cache
def matched(self) -> Optional[int]:
    """Gets the number of products matched the query.

    This method is cached, so it is safe to call it multiple times.

    Returns:
        The number of products matched the query.
    """
    response = self._request({"count": 0})
    return response.get("properties", {}).get("numberOfRecordsMatched")

products(convert_types=True)

Yields products matched the query.

Parameters:

Name Type Description Default
convert_types bool

Whether to convert string values in properties into int, float, bool, or None.

 True

Yields:

Type Description
Iterator[Product]

A Product instance.
Source code in gportal/search.py 
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
def products(self, convert_types: bool = True) -> Iterator[Product]:
    """Yields products matched the query.

    Args:
        convert_types: Whether to convert string values in properties into int, float, bool, or None.

    Yields:
        A [Product][gportal.product.Product] instance.
    """
    for page in self.pages():
        for product_dict in page.get("features", []):
            yield Product(product_dict, convert_types=convert_types)

pages()

Iterates a search request with pagination.

Yields:

Type Description
Iterator[dict[str, Any]]

A dictionary of GeoJSON FeatureCollection.
Source code in gportal/search.py 
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
def pages(self) -> Iterator[dict[str, Any]]:
    """Iterates a search request with pagination.

    Yields:
        A dictionary of GeoJSON FeatureCollection.
    """
    start_index = 1

    while True:
        page = self._request({"startIndex": start_index})
        yield page

        properties = page["properties"]
        start_index += properties["numberOfRecordsReturned"]

        if start_index > properties["numberOfRecordsMatched"]:
            break