#153: remove unsupported characters from Elasticsearch index name

Author: StijnCaerts

stac_fastapi/elasticsearch/stac_fastapi/elasticsearch/database_logic.py

@@ -24,6 +24,20 @@
 
 COLLECTIONS_INDEX = os.getenv("STAC_COLLECTIONS_INDEX", "collections")
 ITEMS_INDEX_PREFIX = os.getenv("STAC_ITEMS_INDEX_PREFIX", "items_")
+ES_INDEX_NAME_UNSUPPORTED_CHARS = {
+    "\\",
+    "/",
+    "*",
+    "?",
+    '"',
+    "<",
+    ">",
+    "|",
+    " ",
+    ",",
+    "#",
+    ":",
+}
 
 DEFAULT_INDICES = f"*,-*kibana*,-{COLLECTIONS_INDEX}"
 
@@ -136,7 +150,7 @@ def index_by_collection_id(collection_id: str) -> str:
     Returns:
         str: The index name derived from the collection id.
     """
-    return f"{ITEMS_INDEX_PREFIX}{collection_id}"
+    return f"{ITEMS_INDEX_PREFIX}{''.join(c for c in collection_id.lower() if c not in ES_INDEX_NAME_UNSUPPORTED_CHARS)}"
 
 
 def indices(collection_ids: Optional[List[str]]) -> str:
@@ -152,7 +166,7 @@ def indices(collection_ids: Optional[List[str]]) -> str:
     if collection_ids is None:
         return DEFAULT_INDICES
     else:
-        return ",".join([f"{ITEMS_INDEX_PREFIX}{c.strip()}" for c in collection_ids])
+        return ",".join([index_by_collection_id(c) for c in collection_ids])
 
 
 async def create_collection_index() -> None:

stac_fastapi/elasticsearch/tests/resources/test_item.py

@@ -75,6 +75,18 @@ async def test_create_item_missing_collection(app_client, ctx):
     assert resp.status_code == 404
 
 
+async def test_create_uppercase_collection_with_item(app_client, ctx, txn_client):
+    """Test creation of a collection and item with uppercase collection ID (transactions extension)"""
+    collection_id = "UPPERCASE"
+    ctx.item["collection"] = collection_id
+    ctx.collection["id"] = collection_id
+    resp = await app_client.post("/collections", json=ctx.collection)
+    assert resp.status_code == 200
+    await refresh_indices(txn_client)
+    resp = await app_client.post(f"/collections/{collection_id}/items", json=ctx.item)
+    assert resp.status_code == 200
+
+
 async def test_update_item_already_exists(app_client, ctx):
     """Test updating an item which already exists (transactions extension)"""