Merge branch 'sort-item-collection' into collections-items-fields-extension

# Conflicts:
#	CHANGELOG.md
#	stac_fastapi/core/stac_fastapi/core/core.py

Author: z-mrozu

CHANGELOG.md

@@ -10,14 +10,21 @@ and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0.
 
 ### Added
 
+- Added the `ENV_MAX_LIMIT` environment variable to SFEOS, allowing overriding of the `MAX_LIMIT`, which controls the `?limit` parameter for returned items and STAC collections. [#434](https://github.com/stac-utils/stac-fastapi-elasticsearch-opensearch/pull/434)
+- Sort, Query, and Filter extension and functionality to the item collection route. [#437](https://github.com/stac-utils/stac-fastapi-elasticsearch-opensearch/pull/437)
 - Added Fields Extension implementation for the `/collections/{collection_id}/aggregations` endpoint. [#436](https://github.com/stac-utils/stac-fastapi-elasticsearch-opensearch/pull/436)
 
 ### Changed
 
-- Changed assets serialization to prevent mapping explosion while allowing asset inforamtion to be indexed. [#341](https://github.com/stac-utils/stac-fastapi-elasticsearch-opensearch/pull/341)
+- Changed assets serialization to prevent mapping explosion while allowing asset information to be indexed. [#341](https://github.com/stac-utils/stac-fastapi-elasticsearch-opensearch/pull/341)
+- Simplified the item_collection function in core.py, moving the request to the get_search function. [#437](https://github.com/stac-utils/stac-fastapi-elasticsearch-opensearch/pull/437)
+- Updated the `format_datetime_range` function to support milliseconds. [#423](https://github.com/stac-utils/stac-fastapi-elasticsearch-opensearch/pull/423)
+- Blocked the /collections/{collection_id}/bulk_items endpoint when environmental variable ENABLE_DATETIME_INDEX_FILTERING is set to true. [#438](https://github.com/stac-utils/stac-fastapi-elasticsearch-opensearch/pull/438)
 
 ### Fixed
 
+- Fixed issue where sortby was not accepting the default sort, where a + or - was not specified before the field value ie. localhost:8081/collections/{collection_id}/items?sortby=id. [#437](https://github.com/stac-utils/stac-fastapi-elasticsearch-opensearch/pull/437)
+
 ## [v6.2.1] - 2025-09-02
 
 ### Added

README.md

@@ -228,6 +228,7 @@ You can customize additional settings in your `.env` file:
 | `DATABASE_REFRESH`           | Controls whether database operations refresh the index immediately after changes. If set to `true`, changes will be immediately searchable. If set to `false`, changes may not be immediately visible but can improve performance for bulk operations. If set to `wait_for`, changes will wait for the next refresh cycle to become visible. | `false` | Optional |
 | `ENABLE_TRANSACTIONS_EXTENSIONS` | Enables or disables the Transactions and Bulk Transactions API extensions. If set to `false`, the POST `/collections` route and related transaction endpoints (including bulk transaction operations) will be unavailable in the API. This is useful for deployments where mutating the catalog via the API should be prevented. | `true` | Optional |
 | `STAC_ITEM_LIMIT` | Sets the environment variable for result limiting to SFEOS for the number of returned items and STAC collections. | `10` | Optional |
+| `ENV_MAX_LIMIT` | Configures the environment variable in SFEOS to override the default `MAX_LIMIT`, which controls the limit parameter for returned items and STAC collections. | `10,000` | Optional |
 
 > [!NOTE]
 > The variables `ES_HOST`, `ES_PORT`, `ES_USE_SSL`, `ES_VERIFY_CERTS` and `ES_TIMEOUT` apply to both Elasticsearch and OpenSearch backends, so there is no need to rename the key names to `OS_` even if you're using OpenSearch.

dockerfiles/Dockerfile.dev.os

@@ -4,11 +4,10 @@ FROM python:3.10-slim
 # update apt pkgs, and install build-essential for ciso8601
 RUN apt-get update && \
     apt-get -y upgrade && \
-    apt-get -y install build-essential && \
+    apt-get -y install build-essential git && \
     apt-get clean && \
     rm -rf /var/lib/apt/lists/*
 
-RUN apt-get -y install git
 # update certs used by Requests
 ENV CURL_CA_BUNDLE=/etc/ssl/certs/ca-certificates.crt
 

stac_fastapi/core/stac_fastapi/core/core.py

@@ -284,106 +284,60 @@ async def get_collection(
     async def item_collection(
         self,
         collection_id: str,
+        request: Request,
         bbox: Optional[BBox] = None,
         datetime: Optional[str] = None,
         limit: Optional[int] = None,
+        sortby: Optional[str] = None,
+        filter_expr: Optional[str] = None,
+        filter_lang: Optional[str] = None,
         token: Optional[str] = None,
+        query: Optional[str] = None,
         **kwargs,
     ) -> stac_types.ItemCollection:
-        """Read items from a specific collection in the database.
+        """List items within a specific collection.
+
+        This endpoint delegates to ``get_search`` under the hood with
+        ``collections=[collection_id]`` so that filtering, sorting and pagination
+        behave identically to the Search endpoints.
 
         Args:
-            collection_id (str): The identifier of the collection to read items from.
-            bbox (Optional[BBox]): The bounding box to filter items by.
-            datetime (Optional[str]): The datetime range to filter items by.
-            limit (int): The maximum number of items to return.
-            token (str): A token used for pagination.
-            request (Request): The incoming request.
+            collection_id (str): ID of the collection to list items from.
+            request (Request): FastAPI Request object.
+            bbox (Optional[BBox]): Optional bounding box filter.
+            datetime (Optional[str]): Optional datetime or interval filter.
+            limit (Optional[int]): Optional page size. Defaults to env ``STAC_ITEM_LIMIT`` when unset.
+            sortby (Optional[str]): Optional sort specification. Accepts repeated values
+                like ``sortby=-properties.datetime`` or ``sortby=+id``. Bare fields (e.g. ``sortby=id``)
+                imply ascending order.
+            token (Optional[str]): Optional pagination token.
+            query (Optional[str]): Optional query string.
+            filter_expr (Optional[str]): Optional filter expression.
+            filter_lang (Optional[str]): Optional filter language.
 
         Returns:
-            ItemCollection: An `ItemCollection` object containing the items from the specified collection that meet
-                the filter criteria and links to various resources.
+            ItemCollection: Feature collection with items, paging links, and counts.
 
         Raises:
-            HTTPException: If the specified collection is not found.
-            Exception: If any error occurs while reading the items from the database.
+            HTTPException: 404 if the collection does not exist.
         """
-        request: Request = kwargs["request"]
-        token = request.query_params.get("token")
-
-        base_url = str(request.base_url)
-
-        collection = await self.get_collection(
-            collection_id=collection_id, request=request
-        )
-        collection_id = collection.get("id")
-        if collection_id is None:
-            raise HTTPException(status_code=404, detail="Collection not found")
-
-        search = self.database.make_search()
-        search = self.database.apply_collections_filter(
-            search=search, collection_ids=[collection_id]
-        )
-
         try:
-            search, datetime_search = self.database.apply_datetime_filter(
-                search=search, datetime=datetime
-            )
-        except (ValueError, TypeError) as e:
-            # Handle invalid interval formats if return_date fails
-            msg = f"Invalid interval format: {datetime}, error: {e}"
-            logger.error(msg)
-            raise HTTPException(status_code=400, detail=msg)
-
-        if bbox:
-            bbox = [float(x) for x in bbox]
-            if len(bbox) == 6:
-                bbox = [bbox[0], bbox[1], bbox[3], bbox[4]]
-
-            search = self.database.apply_bbox_filter(search=search, bbox=bbox)
+            await self.get_collection(collection_id=collection_id, request=request)
+        except Exception:
+            raise HTTPException(status_code=404, detail="Collection not found")
 
-        limit = int(request.query_params.get("limit", os.getenv("STAC_ITEM_LIMIT", 10)))
-        items, maybe_count, next_token = await self.database.execute_search(
-            search=search,
+        # Delegate directly to GET search for consistency
+        return await self.get_search(
+            request=request,
+            collections=[collection_id],
+            bbox=bbox,
+            datetime=datetime,
             limit=limit,
-            sort=None,
             token=token,
-            collection_ids=[collection_id],
-            datetime_search=datetime_search,
-        )
-
-        fields = request.query_params.get("fields")
-        if fields and self.extension_is_enabled("FieldsExtension"):
-            fields = fields.split(",")
-            includes, excludes = set(), set()
-            for field in fields:
-                if field[0] == "-":
-                    excludes.add(field[1:])
-                else:
-                    includes.add(field[1:] if field[0] in "+ " else field)
-
-            items = [
-                filter_fields(
-                    self.item_serializer.db_to_stac(item, base_url=base_url),
-                    includes,
-                    excludes,
-                )
-                for item in items
-            ]
-        else:
-            items = [
-                self.item_serializer.db_to_stac(item, base_url=base_url)
-                for item in items
-            ]
-
-        links = await PagingLinks(request=request, next=next_token).get_links()
-
-        return stac_types.ItemCollection(
-            type="FeatureCollection",
-            features=items,
-            links=links,
-            numReturned=len(items),
-            numMatched=maybe_count,
+            sortby=sortby,
+            query=query,
+            filter_expr=filter_expr,
+            filter_lang=filter_lang,
         )
 
     async def get_item(
@@ -449,6 +403,7 @@ async def get_search(
             HTTPException: If any error occurs while searching the catalog.
         """
         limit = int(request.query_params.get("limit", os.getenv("STAC_ITEM_LIMIT", 10)))
+
         base_args = {
             "collections": collections,
             "ids": ids,
@@ -466,10 +421,18 @@ async def get_search(
             base_args["intersects"] = orjson.loads(unquote_plus(intersects))
 
         if sortby:
-            base_args["sortby"] = [
-                {"field": sort[1:], "direction": "desc" if sort[0] == "-" else "asc"}
-                for sort in sortby
-            ]
+            parsed_sort = []
+            for raw in sortby:
+                if not isinstance(raw, str):
+                    continue
+                s = raw.strip()
+                if not s:
+                    continue
+                direction = "desc" if s[0] == "-" else "asc"
+                field = s[1:] if s and s[0] in "+-" else s
+                parsed_sort.append({"field": field, "direction": direction})
+            if parsed_sort:
+                base_args["sortby"] = parsed_sort
 
         if filter_expr:
             base_args["filter_lang"] = "cql2-json"
@@ -546,13 +509,15 @@ async def post_search(
 
             search = self.database.apply_bbox_filter(search=search, bbox=bbox)
 
-        if search_request.intersects:
+        if hasattr(search_request, "intersects") and getattr(
+            search_request, "intersects"
+        ):
             search = self.database.apply_intersects_filter(
-                search=search, intersects=search_request.intersects
+                search=search, intersects=getattr(search_request, "intersects")
             )
 
-        if search_request.query:
-            for field_name, expr in search_request.query.items():
+        if hasattr(search_request, "query") and getattr(search_request, "query"):
+            for field_name, expr in getattr(search_request, "query").items():
                 field = "properties__" + field_name
                 for op, value in expr.items():
                     # Convert enum to string
@@ -561,9 +526,14 @@ async def post_search(
                         search=search, op=operator, field=field, value=value
                     )
 
-        # only cql2_json is supported here
+        # Apply CQL2 filter (support both 'filter_expr' and canonical 'filter')
+        cql2_filter = None
         if hasattr(search_request, "filter_expr"):
             cql2_filter = getattr(search_request, "filter_expr", None)
+        if cql2_filter is None and hasattr(search_request, "filter"):
+            cql2_filter = getattr(search_request, "filter", None)
+
+        if cql2_filter is not None:
             try:
                 search = await self.database.apply_cql2_filter(search, cql2_filter)
             except Exception as e:
@@ -581,19 +551,23 @@ async def post_search(
                 )
 
         sort = None
-        if search_request.sortby:
-            sort = self.database.populate_sort(search_request.sortby)
+        if hasattr(search_request, "sortby") and getattr(search_request, "sortby"):
+            sort = self.database.populate_sort(getattr(search_request, "sortby"))
 
         limit = 10
         if search_request.limit:
             limit = search_request.limit
 
+        # Use token from the request if the model doesn't define it
+        token_param = getattr(
+            search_request, "token", None
+        ) or request.query_params.get("token")
         items, maybe_count, next_token = await self.database.execute_search(
             search=search,
             limit=limit,
-            token=search_request.token,
+            token=token_param,
             sort=sort,
-            collection_ids=search_request.collections,
+            collection_ids=getattr(search_request, "collections", None),
             datetime_search=datetime_search,
         )
 
@@ -937,7 +911,7 @@ async def delete_collection(self, collection_id: str, **kwargs) -> None:
 
 @attr.s
 class BulkTransactionsClient(BaseBulkTransactionsClient):
-    """A client for posting bulk transactions to a Postgres database.
+    """A client for posting bulk transactions.
 
     Attributes:
         session: An instance of `Session` to use for database connection.
@@ -985,6 +959,13 @@ def bulk_item_insert(
             A string indicating the number of items successfully added.
         """
         request = kwargs.get("request")
+
+        if os.getenv("ENABLE_DATETIME_INDEX_FILTERING"):
+            raise HTTPException(
+                status_code=400,
+                detail="The /collections/{collection_id}/bulk_items endpoint is invalid when ENABLE_DATETIME_INDEX_FILTERING is set to true. Try using the /collections/{collection_id}/items endpoint.",
+            )
+
         if request:
             base_url = str(request.base_url)
         else:

stac_fastapi/core/stac_fastapi/core/datetime_utils.py

@@ -17,17 +17,20 @@ def format_datetime_range(date_str: str) -> str:
     """
 
     def normalize(dt):
+        """Normalize datetime string and preserve millisecond precision."""
         dt = dt.strip()
         if not dt or dt == "..":
             return ".."
         dt_obj = rfc3339_str_to_datetime(dt)
         dt_utc = dt_obj.astimezone(timezone.utc)
-        return dt_utc.strftime("%Y-%m-%dT%H:%M:%SZ")
+        return dt_utc.isoformat(timespec="milliseconds").replace("+00:00", "Z")
 
     if not isinstance(date_str, str):
         return "../.."
+
     if "/" not in date_str:
         return f"{normalize(date_str)}/{normalize(date_str)}"
+
     try:
         start, end = date_str.split("/", 1)
     except Exception:

stac_fastapi/core/stac_fastapi/core/utilities.py

@@ -10,7 +10,15 @@
 
 from stac_fastapi.types.stac import Item
 
-MAX_LIMIT = 10000
+
+def get_max_limit():
+    """
+    Retrieve a MAX_LIMIT value from an environment variable.
+
+    Returns:
+        int: The int value parsed from the environment variable.
+    """
+    return int(os.getenv("ENV_MAX_LIMIT", 10000))
 
 
 def get_bool_env(name: str, default: Union[bool, str] = False) -> bool: