Module stac_fastapi.core.utilities

Module for geospatial processing functions.

This module contains functions for transforming geospatial coordinates, such as converting bounding boxes to polygon representations.

Variables

MAX_LIMIT

logger

Functions

bbox2polygon

def bbox2polygon(
    b0: float,
    b1: float,
    b2: float,
    b3: float
) -> list[list[list[float]]]

Transform a bounding box represented by its four coordinates b0, b1, b2, and b3 into a polygon.

Parameters:

Name	Type	Description	Default
b0	float	The x-coordinate of the lower-left corner of the bounding box.	None
b1	float	The y-coordinate of the lower-left corner of the bounding box.	None
b2	float	The x-coordinate of the upper-right corner of the bounding box.	None
b3	float	The y-coordinate of the upper-right corner of the bounding box.	None

Returns:

Type	Description
List[List[List[float]]]	A polygon represented as a list of lists of coordinates.

build_bulk_summary

def build_bulk_summary(
    raw_features: list,
    processed_items: list,
    valid_items: list,
    validation_error_count: int,
    conflict_errors: list | None = None,
    other_errors: list | None = None
) -> dict

Build a standardized summary dictionary for bulk operations telemetry.

Aggregates counts from all processing layers (preprocessing, validation, database) to provide a comprehensive view of batch operation results.

Parameters:

Name	Type	Description	Default
raw_features	None	Original input features before any processing.	None
processed_items	None	Items that passed preprocessing.	None
valid_items	None	Items that passed validation.	None
validation_error_count	None	Count of validation errors encountered.	None
conflict_errors	None	List of items that caused database conflicts (duplicates).	None
other_errors	None	List of items that caused other database errors.	None

Returns:

Type	Description
None	Dictionary with telemetry counts for input, processed, valid, skipped, and error items.

count_validation_errors

def count_validation_errors(
    validation_errors: dict[str, list[str]]
) -> int

Count total validation errors across all unique error profiles (DRY Helper).

Aggregates error counts from a validation_errors dictionary that maps error messages to lists of affected item IDs.

Parameters:

Name	Type	Description	Default
validation_errors	None	Dictionary mapping error messages to lists of item IDs.	None

Returns:

Type	Description
None	Total count of validation errors across all error types.

dict_deep_update

def dict_deep_update(
    merge_to: dict[str, typing.Any],
    merge_from: dict[str, typing.Any]
) -> None

Perform a deep update of two dicts.

merge_to is updated in-place with the values from merge_from. merge_from values take precedence over existing values in merge_to.

filter_fields

def filter_fields(
    item: stac_fastapi.types.stac.Item | dict[str, typing.Any],
    include: set[str] | None = None,
    exclude: set[str] | None = None
) -> stac_fastapi.types.stac.Item

Preserve and remove fields as indicated by the fields extension include/exclude sets.

Returns a shallow copy of the Item with the fields filtered.

This will not perform a deep copy; values of the original item will be referenced in the return item.

format_conflict_errors

def format_conflict_errors(
    conflicts: list[dict]
) -> dict[str, str]

Format database conflict errors into a clean dictionary.

Parses conflict error objects and extracts human-readable messages. This function is defined at module-load time to avoid memory reallocation on every API request.

Parameters:

Name	Type	Description	Default
conflicts	None	List of conflict error dictionaries from the database.	None

Returns:

Type	Description
None	Dictionary mapping item IDs to human-readable conflict messages.

get_bool_env

def get_bool_env(
    name: str,
    default: bool | str = False
) -> bool

Retrieve a boolean value from an environment variable.

Parameters:

Name	Type	Description	Default
name	str	The name of the environment variable.	None
default	bool	str	The default value to use if the variable is not set or unrecognized. Defaults to False.

Returns:

Type	Description
bool	The boolean value parsed from the environment variable.

get_excluded_from_items

def get_excluded_from_items(
    obj: dict,
    field_path: str
) -> None

Remove a field from items.

The field is removed in-place from the dictionary if it exists. If any intermediate path does not exist or is not a dictionary, the function returns without making any changes.

get_int_env

def get_int_env(
    name: str,
    default: int = 0
) -> int

Retrieve an integer value from an environment variable.

Parameters:

Name	Type	Description	Default
name	str	The name of the environment variable.	None
default	int	The default value to use if the variable is not set or invalid. Defaults to 0.	0

Returns:

Type	Description
int	The integer value parsed from the environment variable.

queue_items_if_enabled

def queue_items_if_enabled(
    collection_id: str,
    items: dict | list[dict]
) -> str | None

Queue items to Redis if ENABLE_REDIS_QUEUE is set.

Handles both single items and bulk items. Returns a status message if queuing was performed, or None if queuing is disabled.

Parameters:

Name	Type	Description	Default
collection_id	None	The collection ID to queue items for.	None
items	None	Single item dict or list of item dicts to queue.	None

Returns:

Type	Description
None	Status message if items were queued, None if queuing is disabled.

Raises:

Type	Description
Exception	Any exception from the queue manager is propagated.