move shared bbox code to helpers

Author: jonhealy1

stac_fastapi/elasticsearch/stac_fastapi/elasticsearch/database_logic.py

@@ -29,6 +29,7 @@
 )
 from stac_fastapi.sfeos_helpers import filter as filter_module
 from stac_fastapi.sfeos_helpers.database import (
+    apply_collections_bbox_filter_shared,
     apply_free_text_filter_shared,
     apply_intersects_filter_shared,
     create_index_templates_shared,
@@ -317,42 +318,9 @@ async def get_all_collections(
                 raise
 
         # Apply bbox filter if provided
-        if bbox:
-            # Parse bbox if it's a string (from GET requests)
-            if isinstance(bbox, str):
-                try:
-                    bbox = [float(x.strip()) for x in bbox.split(",")]
-                except (ValueError, AttributeError) as e:
-                    logger.error(f"Invalid bbox format: {bbox}, error: {e}")
-                    bbox = None
-
-            if bbox and len(bbox) >= 4:
-                # Extract 2D coordinates (bbox can be 2D [minx, miny, maxx, maxy] or 3D [minx, miny, minz, maxx, maxy, maxz])
-                # For geospatial queries, we discard altitude (z) values
-                minx, miny = bbox[0], bbox[1]
-                if len(bbox) == 4:
-                    # 2D bbox
-                    maxx, maxy = bbox[2], bbox[3]
-                else:
-                    # 3D bbox - extract indices 3,4 for maxx,maxy, discarding altitude at indices 2 (minz) and 5 (maxz)
-                    maxx, maxy = bbox[3], bbox[4]
-
-                # Convert bbox to a polygon for geo_shape query
-                bbox_polygon = {
-                    "type": "Polygon",
-                    "coordinates": bbox2polygon(minx, miny, maxx, maxy),
-                }
-                # Add geo_shape query to filter collections by bbox_shape field
-                query_parts.append(
-                    {
-                        "geo_shape": {
-                            "bbox_shape": {
-                                "shape": bbox_polygon,
-                                "relation": "intersects",
-                            }
-                        }
-                    }
-                )
+        bbox_filter = apply_collections_bbox_filter_shared(bbox)
+        if bbox_filter:
+            query_parts.append(bbox_filter)
 
         # Combine all query parts with AND logic if there are multiple
         datetime_filter = None

stac_fastapi/opensearch/stac_fastapi/opensearch/database_logic.py

@@ -29,6 +29,7 @@
 from stac_fastapi.opensearch.config import OpensearchSettings as SyncSearchSettings
 from stac_fastapi.sfeos_helpers import filter as filter_module
 from stac_fastapi.sfeos_helpers.database import (
+    apply_collections_bbox_filter_shared,
     apply_free_text_filter_shared,
     apply_intersects_filter_shared,
     create_index_templates_shared,
@@ -301,42 +302,9 @@ async def get_all_collections(
                 raise
 
         # Apply bbox filter if provided
-        if bbox:
-            # Parse bbox if it's a string (from GET requests)
-            if isinstance(bbox, str):
-                try:
-                    bbox = [float(x.strip()) for x in bbox.split(",")]
-                except (ValueError, AttributeError) as e:
-                    logger.error(f"Invalid bbox format: {bbox}, error: {e}")
-                    bbox = None
-
-            if bbox and len(bbox) >= 4:
-                # Extract 2D coordinates (bbox can be 2D [minx, miny, maxx, maxy] or 3D [minx, miny, minz, maxx, maxy, maxz])
-                # For geospatial queries, we discard altitude (z) values
-                minx, miny = bbox[0], bbox[1]
-                if len(bbox) == 4:
-                    # 2D bbox
-                    maxx, maxy = bbox[2], bbox[3]
-                else:
-                    # 3D bbox - extract indices 3,4 for maxx,maxy, discarding altitude at indices 2 (minz) and 5 (maxz)
-                    maxx, maxy = bbox[3], bbox[4]
-
-                # Convert bbox to a polygon for geo_shape query
-                bbox_polygon = {
-                    "type": "Polygon",
-                    "coordinates": bbox2polygon(minx, miny, maxx, maxy),
-                }
-                # Add geo_shape query to filter collections by bbox_shape field
-                query_parts.append(
-                    {
-                        "geo_shape": {
-                            "bbox_shape": {
-                                "shape": bbox_polygon,
-                                "relation": "intersects",
-                            }
-                        }
-                    }
-                )
+        bbox_filter = apply_collections_bbox_filter_shared(bbox)
+        if bbox_filter:
+            query_parts.append(bbox_filter)
 
         # Combine all query parts with AND logic if there are multiple
         datetime_filter = None

stac_fastapi/sfeos_helpers/stac_fastapi/sfeos_helpers/database/__init__.py

@@ -42,6 +42,7 @@
 )
 from .mapping import get_queryables_mapping_shared
 from .query import (
+    apply_collections_bbox_filter_shared,
     apply_free_text_filter_shared,
     apply_intersects_filter_shared,
     populate_sort_shared,
@@ -59,6 +60,7 @@
     # Query operations
     "apply_free_text_filter_shared",
     "apply_intersects_filter_shared",
+    "apply_collections_bbox_filter_shared",
     "populate_sort_shared",
     # Mapping operations
     "get_queryables_mapping_shared",

stac_fastapi/sfeos_helpers/stac_fastapi/sfeos_helpers/database/query.py

@@ -3,8 +3,10 @@
 This module provides functions for building and manipulating Elasticsearch/OpenSearch queries.
 """
 
-from typing import Any, Dict, List, Optional
+import logging
+from typing import Any, Dict, List, Optional, Union
 
+from stac_fastapi.core.utilities import bbox2polygon
 from stac_fastapi.sfeos_helpers.mappings import Geometry
 
 ES_MAX_URL_LENGTH = 4096
@@ -66,6 +68,78 @@ def apply_intersects_filter_shared(
     }
 
 
+def apply_collections_bbox_filter_shared(
+    bbox: Union[str, List[float], None]
+) -> Optional[Dict[str, Dict]]:
+    """Create a geo_shape filter for collections bbox search.
+
+    This function handles bbox parsing from both GET requests (string format) and POST requests
+    (list format), and constructs a geo_shape query for filtering collections by their bbox_shape field.
+
+    Args:
+        bbox: The bounding box parameter. Can be:
+            - A string of comma-separated coordinates (from GET requests)
+            - A list of floats [minx, miny, maxx, maxy] for 2D bbox
+            - None if no bbox filter is provided
+
+    Returns:
+        Optional[Dict[str, Dict]]: A dictionary containing the geo_shape filter configuration
+            that can be used with Elasticsearch/OpenSearch queries, or None if bbox is invalid.
+            Example return value:
+            {
+                "geo_shape": {
+                    "bbox_shape": {
+                        "shape": {
+                            "type": "Polygon",
+                            "coordinates": [[[minx, miny], [maxx, miny], [maxx, maxy], [minx, maxy], [minx, miny]]]
+                        },
+                        "relation": "intersects"
+                    }
+                }
+            }
+
+    Notes:
+        - This function is specifically for filtering collections by their spatial extent
+        - It queries the bbox_shape field (not the geometry field used for items)
+        - The bbox is expected to be 2D (4 values) after any 3D to 2D conversion in the API layer
+    """
+    logger = logging.getLogger(__name__)
+
+    if not bbox:
+        return None
+
+    # Parse bbox if it's a string (from GET requests)
+    if isinstance(bbox, str):
+        try:
+            bbox = [float(x.strip()) for x in bbox.split(",")]
+        except (ValueError, AttributeError) as e:
+            logger.error(f"Invalid bbox format: {bbox}, error: {e}")
+            return None
+
+    if not bbox or len(bbox) != 4:
+        if bbox:
+            logger.warning(
+                f"bbox has incorrect number of coordinates (length={len(bbox)}), expected 4 (2D bbox)"
+            )
+        return None
+
+    # Convert bbox to a polygon for geo_shape query
+    bbox_polygon = {
+        "type": "Polygon",
+        "coordinates": bbox2polygon(bbox[0], bbox[1], bbox[2], bbox[3]),
+    }
+
+    # Return geo_shape query for bbox_shape field
+    return {
+        "geo_shape": {
+            "bbox_shape": {
+                "shape": bbox_polygon,
+                "relation": "intersects",
+            }
+        }
+    }
+
+
 def populate_sort_shared(sortby: List) -> Optional[Dict[str, Dict[str, str]]]:
     """Create a sort configuration for Elasticsearch/OpenSearch queries.