feat: Add hidden item filtering with HIDE_ITEM_PATH

Adds ability for item hiding via HIDE_ITEM_PATH env var which contains the path to
variable on item level specifying if item should be hidden or not. Hidden items are
excluded from search results and get_one_item() returns NotFoundError for hidden items.

- Updated execute_search() to filter hidden items from search/count queries
- Modified get_one_item() to respect hidden status
- Uses add_hidden_filter() utility for consistent filtering

Author: Yuri Zmytrakov

stac_fastapi/elasticsearch/stac_fastapi/elasticsearch/database_logic.py

@@ -53,6 +53,7 @@
     add_collections_to_body,
 )
 from stac_fastapi.sfeos_helpers.database.utils import (
+    add_hidden_filter,
     merge_to_operations,
     operations_to_script,
 )
@@ -406,12 +407,22 @@ async def get_one_item(self, collection_id: str, item_id: str) -> Dict:
         Notes:
             The Item is retrieved from the Elasticsearch database using the `client.get` method,
             with the index for the Collection as the target index and the combined `mk_item_id` as the document id.
+            Item is hidden if hide_item_path is configured via env var.
         """
         try:
+            base_query = {"term": {"_id": mk_item_id(item_id, collection_id)}}
+
+            HIDE_ITEM_PATH = os.getenv("HIDE_ITEM_PATH", None)
+
+            if HIDE_ITEM_PATH:
+                query = add_hidden_filter(base_query, HIDE_ITEM_PATH)
+            else:
+                query = base_query
+
             response = await self.client.search(
                 index=index_alias_by_collection_id(collection_id),
                 body={
-                    "query": {"term": {"_id": mk_item_id(item_id, collection_id)}},
+                    "query": query,
                     "size": 1,
                 },
             )
@@ -854,6 +865,10 @@ async def execute_search(
 
         size_limit = min(limit + 1, max_result_window)
 
+        HIDE_ITEM_PATH = os.getenv("HIDE_ITEM_PATH", None)
+        if HIDE_ITEM_PATH and query:
+            query = add_hidden_filter(query, HIDE_ITEM_PATH)
+
         search_task = asyncio.create_task(
             self.client.search(
                 index=index_param,
@@ -865,11 +880,18 @@ async def execute_search(
             )
         )
 
+        # Apply hidden filter to count query as well
+        count_query = search.to_dict(count=True)
+        if HIDE_ITEM_PATH and "query" in count_query:
+            count_query["query"] = add_hidden_filter(
+                count_query["query"], HIDE_ITEM_PATH
+            )
+
         count_task = asyncio.create_task(
             self.client.count(
                 index=index_param,
                 ignore_unavailable=ignore_unavailable,
-                body=search.to_dict(count=True),
+                body=count_query,
             )
         )
 

stac_fastapi/opensearch/stac_fastapi/opensearch/database_logic.py

@@ -49,6 +49,7 @@
     add_collections_to_body,
 )
 from stac_fastapi.sfeos_helpers.database.utils import (
+    add_hidden_filter,
     merge_to_operations,
     operations_to_script,
 )
@@ -406,12 +407,22 @@ async def get_one_item(self, collection_id: str, item_id: str) -> Dict:
         Notes:
             The Item is retrieved from the Opensearch database using the `client.get` method,
             with the index for the Collection as the target index and the combined `mk_item_id` as the document id.
+            Item is hidden if hide_item_path is configured via env var.
         """
         try:
+            base_query = {"term": {"_id": mk_item_id(item_id, collection_id)}}
+
+            HIDE_ITEM_PATH = os.getenv("HIDE_ITEM_PATH", None)
+
+            if HIDE_ITEM_PATH:
+                query = add_hidden_filter(base_query, HIDE_ITEM_PATH)
+            else:
+                query = base_query
+
             response = await self.client.search(
                 index=index_alias_by_collection_id(collection_id),
                 body={
-                    "query": {"term": {"_id": mk_item_id(item_id, collection_id)}},
+                    "query": query,
                     "size": 1,
                 },
             )
@@ -846,7 +857,11 @@ async def execute_search(
             index_param = ITEM_INDICES
             query = add_collections_to_body(collection_ids, query)
 
-        if query:
+        HIDE_ITEM_PATH = os.getenv("HIDE_ITEM_PATH", None)
+
+        if HIDE_ITEM_PATH:
+            search_body["query"] = add_hidden_filter(query, HIDE_ITEM_PATH)
+        elif query:
             search_body["query"] = query
 
         search_after = None
@@ -871,11 +886,16 @@ async def execute_search(
             )
         )
 
+        # Ensure hidden item is not counted
+        count_query = search.to_dict(count=True)
+        if "query" in count_query:
+            count_query["query"] = add_hidden_filter(count_query["query"])
+
         count_task = asyncio.create_task(
             self.client.count(
                 index=index_param,
                 ignore_unavailable=ignore_unavailable,
-                body=search.to_dict(count=True),
+                body=count_query,
             )
         )
 

stac_fastapi/sfeos_helpers/stac_fastapi/sfeos_helpers/database/utils.py

@@ -5,7 +5,7 @@
 """
 
 import logging
-from typing import Any, Dict, List, Union
+from typing import Any, Dict, List, Optional, Union
 
 from stac_fastapi.core.utilities import bbox2polygon, get_bool_env
 from stac_fastapi.extensions.core.transaction.request import (
@@ -361,3 +361,35 @@ def operations_to_script(operations: List, create_nest: bool = False) -> Dict:
         "lang": "painless",
         "params": params,
     }
+
+
+def add_hidden_filter(
+    query: Optional[Dict[str, Any]] = None, hide_item_path: Optional[str] = None
+) -> Dict[str, Any]:
+    """Add hidden filter to a query to exclude hidden items.
+
+    Args:
+        query: Optional Elasticsearch query to combine with hidden filter
+        hide_item_path: Path to the hidden field (e.g., "properties._private.hidden")
+                       If None or empty, return original query (no filtering)
+
+    Returns:
+        Query with hidden filter applied
+    """
+    if not hide_item_path:
+        return query or {"match_all": {}}
+
+    hidden_filter = {
+        "bool": {
+            "should": [
+                {"term": {hide_item_path: False}},
+                {"bool": {"must_not": {"exists": {"field": hide_item_path}}}},
+            ],
+            "minimum_should_match": 1,
+        }
+    }
+
+    if query:
+        return {"bool": {"must": [query, hidden_filter]}}
+    else:
+        return hidden_filter

stac_fastapi/tests/resources/test_item.py

@@ -1163,3 +1163,59 @@ async def test_search_datetime_with_null_datetime(
         await txn_client.delete_collection(test_collection["id"])
     except Exception as e:
         logger.warning(f"Failed to delete collection: {e}")
+
+
+@pytest.mark.asyncio
+async def test_hidden_item_true(app_client, txn_client, load_test_data):
+    """Test item with hidden=true is filtered out."""
+
+    os.environ["HIDE_ITEM_PATH"] = "properties._private.hidden"
+
+    test_collection = load_test_data("test_collection.json")
+    test_collection["id"] = "test-collection-hidden-true"
+    await create_collection(txn_client, collection=test_collection)
+    test_item = load_test_data("test_item.json")
+    test_item["collection"] = test_collection["id"]
+    test_item["id"] = "hidden-item-true"
+    test_item["properties"]["_private"] = {"hidden": True}
+
+    await create_item(txn_client, test_item)
+
+    resp = await app_client.get("/search", params={"ids": test_item["id"]})
+    assert resp.status_code == 200
+    result = resp.json()
+    assert len(result["features"]) == 0
+
+    resp = await app_client.get(
+        f"/collections/{test_item['collection']}/items/{test_item['id']}"
+    )
+    assert resp.status_code == 404
+
+
+@pytest.mark.asyncio
+async def test_hidden_item_false(app_client, txn_client, load_test_data, monkeypatch):
+    """Test that item with hidden=false is not filtered out."""
+
+    os.environ["HIDE_ITEM_PATH"] = "properties._private.hidden"
+
+    test_collection = load_test_data("test_collection.json")
+    test_collection["id"] = "test-collection-hidden-false"
+    await create_collection(txn_client, collection=test_collection)
+
+    test_item = load_test_data("test_item.json")
+    test_item["id"] = "hidden-item-false"
+    test_item["collection"] = test_collection["id"]
+    test_item["properties"]["_private"] = {"hidden": False}
+
+    await create_item(txn_client, test_item)
+    await refresh_indices(txn_client)
+
+    resp = await app_client.get("/search", params={"ids": test_item["id"]})
+    assert resp.status_code == 200
+    result = resp.json()
+    assert len(result["features"]) == 1
+
+    resp = await app_client.get(
+        f"/collections/{test_item['collection']}/items/{test_item['id']}"
+    )
+    assert resp.status_code == 200