class CatalogSearchClient(CoreCrudClient):
"""Support 3D search bboxes over 2D item geometries at elevation zero.
Adapt only the validated model sent to pgSTAC. The original HTTP request
remains available to upstream GET/POST handlers for pagination links.
This implements the STAC rule for 2D geometries, not volumetric item search.
"""
async def _search_base(self, search_request: PgstacSearch, request: Request) -> ItemCollection:
"""Handle altitude and degenerate bboxes before invoking pgSTAC."""
bbox = search_request.bbox
if bbox is None:
return await super()._search_base(search_request, request)
# The request model already validates length and latitude/elevation order.
# NaN can pass comparison-based bounds checks, so check finiteness first.
if not all(isfinite(value) for value in bbox):
raise HTTPException(HTTP_400_BAD_REQUEST, "Invalid bbox: coordinates must be finite numbers.")
west, south = bbox[:2]
# STAC 3D order is [west, south, min_elevation, east, north, max_elevation].
east, north = bbox[3:5] if len(bbox) == 6 else bbox[2:]
if not (-180 <= west <= 180 and -180 <= east <= 180 and -90 <= south <= 90 and -90 <= north <= 90):
raise HTTPException(HTTP_400_BAD_REQUEST, "Invalid bbox: coordinates must be within WGS84 bounds.")
if len(bbox) == 6 and not bbox[2] <= 0 <= bbox[5]:
# An interval excluding zero cannot intersect a 2D item geometry.
return {
"type": "FeatureCollection",
"features": [],
"links": await PagingLinks(request=request).get_links(),
"numberMatched": 0,
"numberReturned": 0,
}
if west < east and south < north:
# Keep the native pgSTAC bbox path for ordinary rectangles.
if len(bbox) == 6:
# Zero is inside the elevation range here. Projecting to 2D avoids
# pgSTAC's PolyhedralSurface conversion while preserving matches for 2D items.
search_request = search_request.model_copy(deep=True, update={"bbox": (west, south, east, north)})
return await super()._search_base(search_request, request)
geometry: Point | LineString | Polygon | MultiPoint | MultiLineString | MultiPolygon | GeometryCollection
if west > east:
# Split antimeridian-crossing boxes instead of reversing longitude bounds.
western = bbox_to_geometry(west, south, 180, north)
eastern = bbox_to_geometry(-180, south, east, north)
if isinstance(western, Polygon) and isinstance(eastern, Polygon):
geometry = MultiPolygon(type="MultiPolygon", coordinates=[western.coordinates, eastern.coordinates])
elif isinstance(western, LineString) and isinstance(eastern, LineString):
geometry = MultiLineString(
type="MultiLineString",
coordinates=[western.coordinates, eastern.coordinates],
)
elif isinstance(western, Point) and isinstance(eastern, Point):
geometry = MultiPoint(type="MultiPoint", coordinates=[western.coordinates, eastern.coordinates])
else:
# A bound exactly at +/-180 can collapse one half to a line or point;
# mixed geometry types need a GeometryCollection.
geometry = GeometryCollection(type="GeometryCollection", geometries=[western, eastern])
else:
geometry = bbox_to_geometry(west, south, east, north)
# Copy the model without reprocessing existing filters/extensions. The
# replacement geometry is validated above by the GeoJSON models.
# Clear bbox because STAC makes bbox and intersects mutually exclusive.
backend_request = search_request.model_copy(deep=True, update={"bbox": None, "intersects": geometry})
return await super()._search_base(backend_request, request)