Initial implementation of metadata indexing (#173)

* Initial implementation of metadata indexing and also updated syntax for elasticsearch dependency injection

* fixes

* metadata search working

* Enabling metadata to search from UI

* move MongoDBRef

* fixing coroutine error related to async issue

* fixing test case

* adding await to async method

* fixing patch_metadata update

* fixing #201 issue

* fix to metadata insert_record

* Added delete metadata index to delete-data.sh script.

Co-authored-by: Max Burnette <[email protected]>
Co-authored-by: Luigi Marini <[email protected]>

Author: 3 people

backend/app/database/errors.py

@@ -8,7 +8,7 @@
 from app.config import settings
 from app.mongo import crete_mongo_indexes
 from app.models.errors import Error
-from app.models.metadata import MongoDBRef
+from app.models.mongomodel import MongoDBRef
 
 logger = logging.getLogger(__name__)
 

backend/app/dependencies.py

@@ -63,5 +63,6 @@ def get_rabbitmq() -> BlockingChannel:
     return channel
 
 
-def get_elasticsearchclient():
-    return connect_elasticsearch()
+async def get_elasticsearchclient():
+    es = await connect_elasticsearch()
+    return es

backend/app/main.py

@@ -152,6 +152,9 @@ async def startup_elasticsearch():
     create_index(
         es, "dataset", settings.elasticsearch_setting, indexSettings.dataset_mappings
     )
+    create_index(
+        es, "metadata", settings.elasticsearch_setting, indexSettings.metadata_mappings
+    )
 
 
 @app.on_event("shutdown")

backend/app/models/errors.py

@@ -2,8 +2,7 @@
 from typing import Optional
 from pydantic import Field
 
-from app.models.mongomodel import MongoModel
-from app.models.metadata import MongoDBRef
+from app.models.mongomodel import MongoModel, MongoDBRef
 
 
 class ServiceUnreachable(Exception):

backend/app/models/metadata.py

@@ -4,26 +4,24 @@
 from typing import Optional, List, Union
 from enum import Enum
 
+from elasticsearch import Elasticsearch
 from bson import ObjectId
 from bson.dbref import DBRef
+from fastapi.param_functions import Depends
 from pydantic import Field, validator, BaseModel, create_model
 from fastapi import HTTPException
 from pymongo import MongoClient
 
-from app.models.mongomodel import MongoModel
+from app import dependencies
+from app.models.mongomodel import MongoModel, MongoDBRef
 from app.models.pyobjectid import PyObjectId
 from app.models.users import UserOut
 from app.models.listeners import (
     EventListenerIn,
     LegacyEventListenerIn,
     EventListenerOut,
 )
-
-
-class MongoDBRef(BaseModel):
-    collection: str
-    resource_id: PyObjectId
-    version: Optional[int]
+from app.search.connect import update_record
 
 
 # List of valid types that can be specified for metadata fields
@@ -299,7 +297,9 @@ def deep_update(orig: dict, new: dict):
     return orig
 
 
-async def patch_metadata(metadata: dict, new_entries: dict, db: MongoClient):
+async def patch_metadata(
+    metadata: dict, new_entries: dict, db: MongoClient, es: Elasticsearch
+):
     """Convenience function for updating original metadata contents with new entries."""
     try:
         # TODO: For list-type definitions, should we append to list instead?
@@ -315,6 +315,9 @@ async def patch_metadata(metadata: dict, new_entries: dict, db: MongoClient):
         db["metadata"].replace_one(
             {"_id": metadata["_id"]}, MetadataDB(**metadata).to_mongo()
         )
+        # Update entry to the metadata index
+        doc = {"doc": {"contents": metadata["contents"]}}
+        update_record(es, "metadata", doc, metadata["_id"])
     except Exception as e:
         raise e
     return MetadataOut.from_mongo(metadata)

backend/app/models/mongomodel.py

@@ -1,3 +1,4 @@
+from typing import Optional
 from datetime import datetime
 from bson import ObjectId
 from bson.errors import InvalidId
@@ -61,3 +62,9 @@ def to_mongo(self, **kwargs):
             parsed["_id"] = parsed.pop("id")
 
         return parsed
+
+
+class MongoDBRef(BaseModel):
+    collection: str
+    resource_id: PyObjectId
+    version: Optional[int]

backend/app/routers/datasets.py

@@ -9,6 +9,7 @@
 from typing import List, Optional, Union
 import json
 
+from elasticsearch import Elasticsearch
 import pika
 from bson import ObjectId
 from bson import json_util
@@ -190,10 +191,8 @@ async def save_dataset(
     dataset_in: DatasetIn,
     user=Depends(keycloak_auth.get_current_user),
     db: MongoClient = Depends(dependencies.get_db),
-    es=Depends(dependencies.get_elasticsearchclient),
+    es: Elasticsearch = Depends(dependencies.get_elasticsearchclient),
 ):
-    es = await connect_elasticsearch()
-
     # Check all connection and abort if any one of them is not available
     if db is None or es is None:
         raise HTTPException(status_code=503, detail="Service not available")
@@ -348,7 +347,7 @@ async def patch_dataset(
     dataset_info: DatasetPatch,
     user_id=Depends(get_user),
     db: MongoClient = Depends(dependencies.get_db),
-    es=Depends(dependencies.get_elasticsearchclient),
+    es: Elasticsearch = Depends(dependencies.get_elasticsearchclient),
 ):
     es = await connect_elasticsearch()
 
@@ -390,7 +389,7 @@ async def delete_dataset(
     dataset_id: str,
     db: MongoClient = Depends(dependencies.get_db),
     fs: Minio = Depends(dependencies.get_fs),
-    es=Depends(dependencies.get_elasticsearchclient),
+    es: Elasticsearch = Depends(dependencies.get_elasticsearchclient),
 ):
     es = await connect_elasticsearch()
 

backend/app/routers/elasticsearch.py

@@ -25,8 +25,15 @@ async def search_dataset(request: Request):
     return search_index(es, "dataset", query)
 
 
-@router.post("/file,dataset/_msearch")
-async def search_file_and_dataset(request: Request):
+@router.post("/metadata/_msearch")
+async def search_metadata(request: Request):
     es = await connect_elasticsearch()
     query = await request.body()
-    return search_index(es, ["file", "dataset"], query)
+    return search_index(es, "metadata", query)
+
+
+@router.post("/file,dataset,metadata/_msearch")
+async def search_file_dataset_and_metadata(request: Request):
+    es = await connect_elasticsearch()
+    query = await request.body()
+    return search_index(es, ["file", "dataset", "metadata"], query)

backend/app/routers/files.py

@@ -1,5 +1,7 @@
 import io
 import json
+
+from elasticsearch import Elasticsearch
 import pika
 import mimetypes
 from datetime import datetime
@@ -47,7 +49,7 @@ async def add_file_entry(
     fs: Minio,
     file: Optional[io.BytesIO] = None,
     content_type: Optional[str] = None,
-    es=Depends(dependencies.get_elasticsearchclient),
+    es: Elasticsearch = Depends(dependencies.get_elasticsearchclient),
 ):
     """Insert FileDB object into MongoDB (makes Clowder ID), then Minio (makes version ID), then update MongoDB with
     the version ID from Minio.
@@ -121,7 +123,7 @@ async def remove_file_entry(
     file_id: Union[str, ObjectId],
     db: MongoClient,
     fs: Minio,
-    es=Depends(dependencies.get_elasticsearchclient),
+    es: Elasticsearch = Depends(dependencies.get_elasticsearchclient),
 ):
     """Remove FileDB object into MongoDB, Minio, and associated metadata and version information."""
     # TODO: Deleting individual versions will require updating version_id in mongo, or deleting entire document
@@ -148,7 +150,7 @@ async def update_file(
     db: MongoClient = Depends(dependencies.get_db),
     fs: Minio = Depends(dependencies.get_fs),
     file: UploadFile = File(...),
-    es=Depends(dependencies.get_elasticsearchclient),
+    es: Elasticsearch = Depends(dependencies.get_elasticsearchclient),
 ):
     es = await connect_elasticsearch()
 

backend/app/routers/metadata_datasets.py

@@ -3,6 +3,7 @@
 import os
 from typing import List, Optional
 
+from elasticsearch import Elasticsearch
 from bson import ObjectId
 from fastapi import APIRouter, HTTPException, Depends
 from fastapi import Form
@@ -26,6 +27,7 @@
     patch_metadata,
     MetadataDelete,
 )
+from app.search.connect import insert_record, update_record, delete_document_by_id
 
 router = APIRouter()
 
@@ -80,6 +82,7 @@ async def add_dataset_metadata(
     dataset_id: str,
     user=Depends(get_current_user),
     db: MongoClient = Depends(dependencies.get_db),
+    es: Elasticsearch = Depends(dependencies.get_elasticsearchclient),
 ):
     """Attach new metadata to a dataset. The body must include a contents field with the JSON metadata, and either a
     context JSON-LD object, context_url, or definition (name of a metadata definition) to be valid.
@@ -110,6 +113,18 @@ async def add_dataset_metadata(
         new_metadata = await db["metadata"].insert_one(md.to_mongo())
         found = await db["metadata"].find_one({"_id": new_metadata.inserted_id})
         metadata_out = MetadataOut.from_mongo(found)
+
+        # Add an entry to the metadata index
+        doc = {
+            "resource_id": dataset_id,
+            "resource_type": "dataset",
+            "created": metadata_out.created.utcnow(),
+            "creator": user.email,
+            "contents": metadata_out.contents,
+            "context_url": metadata_out.context_url,
+            "context": metadata_out.context,
+        }
+        insert_record(es, "metadata", doc, metadata_out.id)
         return metadata_out
 
 
@@ -119,6 +134,7 @@ async def replace_dataset_metadata(
     dataset_id: str,
     user=Depends(get_current_user),
     db: MongoClient = Depends(dependencies.get_db),
+    es: Elasticsearch = Depends(dependencies.get_elasticsearchclient),
 ):
     """Update metadata. Any fields provided in the contents JSON will be added or updated in the metadata. If context or
     agent should be changed, use PUT.
@@ -159,6 +175,9 @@ async def replace_dataset_metadata(
             )
             found = await db["metadata"].find_one({"_id": md["_id"]})
             metadata_out = MetadataOut.from_mongo(found)
+            # Update entry to the metadata index
+            doc = {"doc": {"contents": metadata_out["contents"]}}
+            update_record(es, "metadata", doc, metadata_out["_id"])
             return metadata_out
     else:
         raise HTTPException(status_code=404, detail=f"Dataset {dataset_id} not found")
@@ -170,6 +189,7 @@ async def update_dataset_metadata(
     dataset_id: str,
     user=Depends(get_current_user),
     db: MongoClient = Depends(dependencies.get_db),
+    es: Elasticsearch = Depends(dependencies.get_elasticsearchclient),
 ):
     """Update metadata. Any fields provided in the contents JSON will be added or updated in the metadata. If context or
     agent should be changed, use PUT.
@@ -227,7 +247,7 @@ async def update_dataset_metadata(
 
         if (md := await db["metadata"].find_one(query)) is not None:
             # TODO: Refactor this with permissions checks etc.
-            result = await patch_metadata(md, contents, db)
+            result = await patch_metadata(md, contents, db, es)
             return result
         else:
             raise HTTPException(
@@ -278,6 +298,7 @@ async def delete_dataset_metadata(
     dataset_id: str,
     user=Depends(get_current_user),
     db: MongoClient = Depends(dependencies.get_db),
+    es: Elasticsearch = Depends(dependencies.get_elasticsearchclient),
 ):
     if (
         dataset := await db["datasets"].find_one({"_id": ObjectId(dataset_id)})
@@ -327,5 +348,7 @@ async def delete_dataset_metadata(
             raise HTTPException(
                 status_code=404, detail=f"No metadata found with that criteria"
             )
+        # delete from elasticsearch
+        delete_document_by_id(es, "metadata", str(metadata_in.id))
     else:
         raise HTTPException(status_code=404, detail=f"Dataset {dataset_id} not found")