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

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

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

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
HTTPException	If the S3 key does not match the expected pattern, or if multiple buckets are used.

Source code in docs/rs-server/services/catalog/rs_server_catalog/utils.py

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:
        HTTPException: If the S3 key does not match the expected pattern, or if multiple buckets are used.
    """
    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

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

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

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

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,
        )