Skip to content

rs_server_catalog/utils.md

<< Back to index

This library contains functions used in handling the user catalog.

add_prefix_link_landing_page(content, url)

Add the CATALOG_PREFIX to the landing page if it is not present.

Parameters:

Name Type Description Default
content dict

the landing page

 required
url str

the url

 required
Source code in docs/rs-server/services/catalog/rs_server_catalog/utils.py 
128
129
130
131
132
133
134
135
136
137
138
139
140
141
def add_prefix_link_landing_page(content: dict, url: str) -> dict:
    """
    Add the CATALOG_PREFIX to the landing page if it is not present.

    Args:
        content (dict): the landing page
        url (str): the url
    """
    for link in content["links"]:
        if "href" in link and CATALOG_PREFIX not in link["href"]:
            href = link["href"]
            url_size = len(url)
            link["href"] = href[:url_size] + CATALOG_PREFIX + href[url_size:]
    return content

extract_owner_name_from_json_filter(json_filter)

Scans a CQL2 JSON filter and returns the associated owner name if it contains an "owner" property. Owner name must be in this kind of subpart of the filter: "args": [{"property": "owner"}, "toto"]}

Parameters:

Name Type Description Default
json_filter Any

Filter to scan. Expected to be a dictionary (else returns None)

 required

Returns:

Type Description
str | None

str|None: owner name if there is one, None in any other case
Source code in docs/rs-server/services/catalog/rs_server_catalog/utils.py 
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
def extract_owner_name_from_json_filter(json_filter: Any) -> str | None:
    """
    Scans a CQL2 JSON filter and returns the associated owner name if it contains an "owner" property.
    Owner name must be in this kind of subpart of the filter:
        "args": [{"property": "owner"}, "toto"]}

    Args:
        json_filter (Any): Filter to scan. Expected to be a dictionary (else returns None)

    Returns:
        str|None: owner name if there is one, None in any other case
    """
    if not isinstance(json_filter, dict):
        return None

    if "args" in json_filter and isinstance(json_filter["args"], list):
        if (
            len(json_filter["args"]) == 2
            and isinstance(json_filter["args"][0], dict)
            and json_filter["args"][0].get("property") == "owner"
        ):
            return json_filter["args"][1]

        for element in json_filter["args"]:
            owner_name = extract_owner_name_from_json_filter(element)
            if owner_name is not None:
                return owner_name

    return None

extract_owner_name_from_text_filter(text_filter)

Scans a CQL2 text filter and returns the associated owner name if it contains an "owner" property. Owner name must be a field in this kind of filter: "width=2500 AND owner='toto'"

Parameters:

Name Type Description Default
text_filter str

Text filter to scan

 required

Returns:

Type Description
str | None

str|None: owner name if there is one, None in any other case
Source code in docs/rs-server/services/catalog/rs_server_catalog/utils.py 
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
def extract_owner_name_from_text_filter(text_filter: str) -> str | None:
    """
    Scans a CQL2 text filter and returns the associated owner name if it contains an "owner" property.
    Owner name must be a field in this kind of filter:
        "width=2500 AND owner='toto'"

    Args:
        text_filter (str): Text filter to scan

    Returns:
        str|None: owner name if there is one, None in any other case
    """
    # Regex to isolate the content of the "owner" field in a text filter
    owner_regex = r"\bowner\s*=\s*['\"]([^'\"]+)['\"]"

    match = re.search(owner_regex, text_filter, re.IGNORECASE)

    if match:
        return match.group(1)
    return None

get_temp_bucket_name(files_s3_key)

Retrieve the temporary bucket name from a list of S3 keys.

Parameters:

Name Type Description Default
files_s3_key list[str]

A list of S3 key strings.

 required

Returns:

Type Description
str | None

str | None: The name of the temporary S3 bucket if valid, otherwise None.

Raises:

Type Description
RuntimeError

If an S3 key does not match the expected s3://bucket/path pattern, or if the keys span more than one bucket.
Source code in docs/rs-server/services/catalog/rs_server_catalog/utils.py 
 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
def get_temp_bucket_name(files_s3_key: list[str]) -> str | None:
    """
    Retrieve the temporary bucket name from a list of S3 keys.

    Args:
        files_s3_key (list[str]): A list of S3 key strings.

    Returns:
        str | None: The name of the temporary S3 bucket if valid, otherwise None.

    Raises:
        RuntimeError: If an S3 key does not match the expected s3://bucket/path pattern,
            or if the keys span more than one bucket.
    """
    if not files_s3_key:
        return None

    bucket_names = set()

    for s3_key in files_s3_key:
        if not is_s3_path(s3_key):
            raise RuntimeError(
                f"The S3 key '{s3_key}' does not match the correct S3 path pattern " "(s3://bucket_name/path/to/obj)",
            )
        # Extract and add the bucket name to the set
        bucket_names.add(s3_key.split("/")[2])

    if len(bucket_names) != 1:
        raise RuntimeError(f"A single temporary S3 bucket should be used in the assets: {bucket_names!r}")

    return bucket_names.pop()

get_token_for_pagination(items_dic)

Used to get the token to be used when calling functions from the stac-fastapi-pgstac object.

Source code in docs/rs-server/services/catalog/rs_server_catalog/utils.py 
114
115
116
117
118
119
120
def get_token_for_pagination(items_dic: dict[Any, Any]):
    """Used to get the token to be used when calling functions from the stac-fastapi-pgstac object."""
    token = None
    for link in items_dic.get("links", []):
        if link.get("rel") == "next":
            token = link.get("href", None)
    return token

headers_minus_content_length(response)

Returns response headers without Content-Length

Source code in docs/rs-server/services/catalog/rs_server_catalog/utils.py 
123
124
125
def headers_minus_content_length(response: Response) -> dict[str, str]:
    """Returns response headers without Content-Length"""
    return {k: v for k, v in response.headers.items() if k.lower() != "content-length"}

is_s3_path(s3_key)

Function to check if a string matches the S3 pattern

Source code in docs/rs-server/services/catalog/rs_server_catalog/utils.py 
74
75
76
77
78
def is_s3_path(s3_key):
    """Function to check if a string matches the S3 pattern"""
    if not isinstance(s3_key, str):
        return False
    return bool(s3_pattern.match(s3_key))

verify_existing_item_from_catalog(method, item, content_id_str, user_collection_str)

Verify if an exisiting item from the catalog may be created or updated

Parameters:

Name Type Description Default
method str

The HTTP method used in the request (e.g., "POST", "PUT", "PATCH").

 required
item dict

The item from the catalog to check.

 required
content_id_str str

The name of the item, used for generating an error message

 required
user_collection_str str

The collection identifier including the user.

 required

Raises:

Type Description
HTTPException

If a POST request is made for an existing item, or if a PUT/PATCH request is made for a non-existent item.
Source code in docs/rs-server/services/catalog/rs_server_catalog/utils.py 
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
def verify_existing_item_from_catalog(method: str, item: dict, content_id_str: str, user_collection_str: str):
    """Verify if an exisiting item from the catalog may be created or updated

    Args:
        method (str): The HTTP method used in the request (e.g., "POST", "PUT", "PATCH").
        item (dict): The item from the catalog to check.
        content_id_str (str): The name of the item, used for generating an error message
        user_collection_str (str): The collection identifier including the user.

    Raises:
        HTTPException: If a POST request is made for an existing item,
                       or if a PUT/PATCH request is made for a non-existent item.
    """

    # Protection for cases where a POST request attempts to add an
    # item with a name that already exists in the database.
    if method == "POST" and item:
        raise HTTPException(
            detail=f"The item {item['id']} " f"already exists in the {user_collection_str} collection",
            status_code=HTTP_409_CONFLICT,
        )
    # Protection for cases where a PUT or PATCH request is made for an item
    # that does not exist in the database.
    if method in {"PUT", "PATCH"} and not item:
        raise HTTPException(
            detail=f"The item {content_id_str} "
            f"does not exist in the {user_collection_str} collection for an update (PUT / PATCH request received)",
            status_code=HTTP_400_BAD_REQUEST,
        )