Extractors -> listeners, trigger listeners on saved feeds automatically (#114)

* Rename extractors to listeners

* Additional model work

* More implementation

* auto-generation of feeds

* listener assignment

* standardizing search object and search criteria

* add delete endpoints

* name cleanup

* refactor how RMQ message built

* standardize some of the submission methods

* resolve merges, cleanup ds metadata

* Update metadata_datasets.py

* Listener -> Event Listener, add docstrings

* fix remaining imports

* docstrings

* get listener name & version as default

* remove version

* Update feeds.py

Author: max-zilla

backend/app/main.py

@@ -25,8 +25,9 @@
     collections,
     authentication,
     keycloak,
-    extractors,
     elasticsearch,
+    listeners,
+    feeds,
 )
 
 # setup loggers
@@ -114,7 +115,13 @@
     dependencies=[Depends(get_current_username)],
 )
 api_router.include_router(
-    extractors.router,
+    listeners.router,
+    prefix="/listeners",
+    tags=["listeners"],
+    dependencies=[Depends(get_current_username)],
+)
+api_router.include_router(
+    listeners.legacy_router,
     prefix="/extractors",
     tags=["extractors"],
     dependencies=[Depends(get_current_username)],
@@ -125,6 +132,12 @@
     tags=["elasticsearch"],
     dependencies=[Depends(get_current_username)],
 )
+api_router.include_router(
+    feeds.router,
+    prefix="/feeds",
+    tags=["feeds"],
+    dependencies=[Depends(get_current_username)],
+)
 api_router.include_router(keycloak.router, prefix="/auth", tags=["auth"])
 app.include_router(api_router, prefix=settings.API_V2_STR)
 

backend/app/models/extractors.py

@@ -0,0 +1,33 @@
+from datetime import datetime
+from pydantic import Field, BaseModel
+from typing import Optional, List, Union
+from app.models.mongomodel import MongoModel
+from app.models.users import UserOut
+from app.models.search import SearchObject
+from app.models.listeners import EventListenerOut, FeedListener
+
+
+class JobFeed(BaseModel):
+    """A Job Feed is a saved set of search criteria with some number of Event Listeners that can be triggered when new
+    resources match the saved search criteria for the Feed."""
+
+    name: str
+    search: SearchObject
+    listeners: List[FeedListener] = []
+
+
+class FeedBase(JobFeed):
+    description: str = ""
+
+
+class FeedIn(JobFeed):
+    pass
+
+
+class FeedDB(JobFeed, MongoModel):
+    author: UserOut
+    updated: datetime = Field(default_factory=datetime.utcnow)
+
+
+class FeedOut(FeedDB):
+    pass

backend/app/models/feeds.py

@@ -0,0 +1,88 @@
+from datetime import datetime
+from pydantic import Field, BaseModel
+from typing import Optional, List, Union
+from app.models.pyobjectid import PyObjectId
+from app.models.mongomodel import MongoModel
+from app.models.users import UserOut
+
+
+class Repository(MongoModel):
+    """Reference to a repository associated with Event Listener/Extractor."""
+
+    repository_type: str = "git"
+    repository_url: str = ""
+
+
+class ExtractorInfo(BaseModel):
+    """Currently for extractor_info JSON from Clowder v1 extractors for use with to /api/extractors endpoint."""
+
+    author: str  # Referring to author of listener script (e.g. name or email), not Clowder user
+    process: dict
+    maturity: str = "Development"
+    contributors: List[str] = []
+    contexts: List[dict] = []
+    repository: List[Repository] = []
+    external_services: List[str] = []
+    libraries: List[str] = []
+    bibtex: List[str] = []
+    default_labels: List[str] = []
+    categories: List[str] = []
+    parameters: List[dict] = []
+
+
+class EventListenerBase(BaseModel):
+    """An Event Listener is the expanded version of v1 Extractors."""
+
+    name: str
+    version: int = 1
+    description: str = ""
+
+
+class EventListenerIn(EventListenerBase):
+    """On submission, minimum info for a listener is name, version and description. Clowder will use name and version to locate queue."""
+
+    pass
+
+
+class LegacyEventListenerIn(ExtractorInfo):
+    """v1 Extractors can submit data formatted as a LegacyEventListener (i.e. v1 format) and it will be converted to a v2 EventListener."""
+
+    name: str
+    version: str = "1.0"
+    description: str = ""
+
+
+class EventListenerDB(EventListenerBase, MongoModel):
+    """EventListeners have a name, version, author, description, and optionally properties where extractor_info will be saved."""
+
+    author: UserOut
+    created: datetime = Field(default_factory=datetime.utcnow)
+    modified: datetime = Field(default_factory=datetime.utcnow)
+    properties: Optional[ExtractorInfo] = None
+
+
+class EventListenerOut(EventListenerDB):
+    pass
+
+
+class FeedListener(BaseModel):
+    """This is a shorthand POST class for associating an existing EventListener with a Feed. The automatic flag determines
+    whether the Feed will automatically send new matches to the Event Listener."""
+
+    listener_id: PyObjectId
+    automatic: bool  # Listeners can trigger automatically or not on a per-feed basis.
+
+
+class EventListenerMessage(BaseModel):
+    """This describes contents of JSON object that is submitted to RabbitMQ for the Event Listeners/Extractors to consume."""
+
+    host: str = "http://127.0.0.1:8000"
+    secretKey: str = "secretKey"
+    retry_count: int = 0
+    resource_type: str = "file"
+    flags: str = ""
+    filename: str
+    fileSize: int
+    id: str
+    datasetId: str
+    token: str

backend/app/models/listeners.py

@@ -13,7 +13,11 @@
 from app.models.mongomodel import MongoModel
 from app.models.pyobjectid import PyObjectId
 from app.models.users import UserOut
-from app.models.extractors import ExtractorIn, ExtractorOut, ExtractorIdentifier
+from app.models.listeners import (
+    EventListenerIn,
+    LegacyEventListenerIn,
+    EventListenerOut,
+)
 
 
 class MongoDBRef(BaseModel):
@@ -57,7 +61,7 @@ class MetadataField(MongoModel):
 
 class MetadataDefinitionBase(MongoModel):
     """This describes a metadata object with a short name and description, predefined set of fields, and context.
-    These provide a shorthand for use by extractors as well as a source for building GUI widgets to add new entries.
+    These provide a shorthand for use by listeners as well as a source for building GUI widgets to add new entries.
 
     Example: {
         "name" : "LatLon",
@@ -164,11 +168,11 @@ def validate_definition(contents: dict, metadata_def: MetadataDefinitionOut):
 
 
 class MetadataAgent(MongoModel):
-    """Describes the user who created a piece of metadata. If extractor is provided, user refers to the user who
-    triggered the extraction."""
+    """Describes the user who created a piece of metadata. If listener is provided, user refers to the user who
+    triggered the job."""
 
     creator: UserOut
-    extractor: Optional[ExtractorOut]
+    listener: Optional[EventListenerOut]
 
 
 class MetadataBase(MongoModel):
@@ -199,7 +203,8 @@ def definition_is_valid(cls, v):
 
 class MetadataIn(MetadataBase):
     file_version: Optional[int]
-    extractor_info: Optional[ExtractorIn]
+    listener: Optional[EventListenerIn]
+    extractor: Optional[LegacyEventListenerIn]
 
 
 class MetadataPatch(MetadataIn):
@@ -225,7 +230,7 @@ class MetadataPatch(MetadataIn):
 class MetadataDelete(MongoModel):
     metadata_id: Optional[str]  # specific metadata ID we are deleting
     definition: Optional[str]
-    extractor_info: Optional[ExtractorIdentifier]
+    listener: Optional[EventListenerIn]
 
 
 class MetadataDB(MetadataBase):

backend/app/models/metadata.py

@@ -0,0 +1,36 @@
+from datetime import datetime
+from pydantic import BaseModel
+from typing import Optional, List
+
+
+# TODO: may eventually be split by index (resource type)
+class SearchIndexContents(BaseModel):
+    """This describes what is indexed in Elasticsearch for a given resource."""
+
+    id: str
+    name: str
+    creator: str  # currently just email
+    created: datetime
+    download: int
+
+
+class SearchCriteria(BaseModel):
+    field: str
+    operator: str = "=="
+    value: str
+
+
+class SearchObject(BaseModel):
+    """This is a way to save a search (i.e. as a Feed).
+
+    Parameters:
+        index_name -- which ES index to search
+        criteria -- some number of field/operator/value tuples describing the search requirements
+        mode -- and/or determines whether all of the criteria must match, or any of them
+        original -- if the user originally performed a string search, their original text entry is preserved here
+    """
+
+    index_name: str
+    criteria: List[SearchCriteria] = []
+    mode: str = "and"  # and / or
+    original: Optional[str]  # original un-parsed search string

backend/app/models/search.py

@@ -3,11 +3,9 @@
 from packaging import version
 from app.config import settings
 from pymongo import MongoClient
-from app.models.extractors import (
-    ExtractorBase,
-    ExtractorIn,
-    ExtractorDB,
-    ExtractorOut,
+from app.models.listeners import (
+    EventListenerDB,
+    EventListenerOut,
 )
 
 
@@ -18,7 +16,7 @@ def callback(ch, method, properties, body):
     extractor_queue = statusBody["queue"]
     extractor_info = statusBody["extractor_info"]
     extractor_name = extractor_info["name"]
-    extractor_db = ExtractorDB(**extractor_info)
+    extractor_db = EventListenerDB(**extractor_info)
     client = MongoClient(settings.MONGODB_URL)
     db = client["clowder2"]
     existing_extractor = db["extractors"].find_one({"name": extractor_queue})
@@ -29,7 +27,7 @@ def callback(ch, method, properties, body):
             new_extractor = db["extractors"].insert_one(extractor_db.to_mongo())
             found = db["extractors"].find_one({"_id": new_extractor.inserted_id})
             removed = db["extractors"].delete_one({"_id": existing_extractor["_id"]})
-            extractor_out = ExtractorOut.from_mongo(found)
+            extractor_out = EventListenerOut.from_mongo(found)
             print(
                 "extractor updated: "
                 + extractor_name
@@ -42,7 +40,7 @@ def callback(ch, method, properties, body):
     else:
         new_extractor = db["extractors"].insert_one(extractor_db.to_mongo())
         found = db["extractors"].find_one({"_id": new_extractor.inserted_id})
-        extractor_out = ExtractorOut.from_mongo(found)
+        extractor_out = EventListenerOut.from_mongo(found)
         print("new extractor registered: " + extractor_name)
         return extractor_out
 

backend/app/rabbitmq/heartbeat_listener_sync.py

@@ -0,0 +1,45 @@
+import json
+import pika
+from fastapi import Request, HTTPException, Depends
+from pymongo import MongoClient
+from bson import ObjectId
+from pika.adapters.blocking_connection import BlockingChannel
+
+from app.keycloak_auth import get_token
+from app import dependencies
+from app.models.files import FileOut
+from app.models.listeners import EventListenerMessage
+
+
+def submit_file_message(
+    file_out: FileOut,
+    queue: str,
+    routing_key: str,
+    parameters: dict,
+    token: str = Depends(get_token),
+    db: MongoClient = Depends(dependencies.get_db),
+    rabbitmq_client: BlockingChannel = Depends(dependencies.get_rabbitmq),
+):
+    # TODO check if extractor is registered
+    msg_body = EventListenerMessage(
+        filename=file_out.name,
+        fileSize=file_out.bytes,
+        id=file_out.id,
+        datasetId=file_out.dataset_id,
+        secretKey=token,
+    )
+
+    rabbitmq_client.queue_bind(
+        exchange="extractors",
+        queue=queue,
+        routing_key=routing_key,
+    )
+    rabbitmq_client.basic_publish(
+        exchange="extractors",
+        routing_key=routing_key,
+        body=json.dumps(msg_body.dict(), ensure_ascii=False),
+        properties=pika.BasicProperties(
+            content_type="application/json", delivery_mode=1
+        ),
+    )
+    return {"message": "testing", "file_id": file_out.id}

backend/app/rabbitmq/listeners.py

@@ -197,7 +197,6 @@ async def save_dataset(
         raise HTTPException(status_code=503, detail="Service not available")
         return
 
-    result = dataset_in.dict()
     dataset_db = DatasetDB(**dataset_in.dict(), author=user)
     new_dataset = await db["datasets"].insert_one(dataset_db.to_mongo())
     found = await db["datasets"].find_one({"_id": new_dataset.inserted_id})