add find endpoint (#66)

* add find endpoint

* update deps

gdrive/export_api.py

@@ -112,3 +112,22 @@ async def survey_upload_response_task(request):
             await upload_file(id)
     except error.ExportError as e:
         log.error(e.args)
+
+
+class FindModel(BaseModel):
+    """
+    Request body format for the `/find` endpoint
+    """
+
+    responseId: str
+    field: str
+    values: list[str]
+    result_field: str | None
+
+
+@router.post("/find")
+async def find(find: FindModel):
+    # for given responseid, find all occurences of
+    result = find.result_field if find.result_field is not None else find.field
+    export_data = export_client.find(find.responseId, find.field, find.values, result)
+    return export_data

gdrive/export_client.py

@@ -171,3 +171,152 @@ def get_qualtrics_response(surveyId: str, responseId: str):
         )
 
     return r.json()
+
+
+def get_all_InteractionIds(responseId):
+    es = OpenSearch(
+        hosts=[{"host": settings.ES_HOST, "port": settings.ES_PORT}], timeout=300
+    )
+
+    # query for all parent flow intraction ids for a given response id
+    query_interactionId = {
+        "size": 500,
+        "query": {
+            "bool": {
+                "must": [
+                    {"match_phrase": {"properties.outcomeType.value": "survey_data"}},
+                    {"match": {"properties.outcomeDescription.value": f"{responseId}"}},
+                ]
+            }
+        },
+        "_source": ["interactionId"],
+    }
+
+    results_interacitonId = es.search(
+        body=json.dumps(query_interactionId), index="_all"
+    )
+
+    interactionIds_match = list(
+        map(
+            lambda res: res["_source"]["interactionId"],
+            results_interacitonId["hits"]["hits"],
+        )
+    )
+
+    subflow_query_1 = list(
+        map(
+            lambda res: {
+                "bool": {
+                    "must": [
+                        {
+                            "match_phrase": {
+                                "parentInteractionProps.parentInteractionId": f'{res["_source"]["interactionId"]}'
+                            }
+                        },
+                        {"exists": {"field": "interactionId"}},
+                    ]
+                }
+            },
+            results_interacitonId["hits"]["hits"],
+        )
+    )
+
+    subflow_query_2 = list(
+        map(
+            lambda res: {
+                "bool": {
+                    "must": [
+                        {
+                            "match_phrase": {
+                                "properties.outcomeDescription.value": f'{res["_source"]["interactionId"]}'
+                            }
+                        },
+                        {"match_phrase": {"properties.outcomeType.value": "parent_id"}},
+                    ]
+                }
+            },
+            results_interacitonId["hits"]["hits"],
+        )
+    )
+
+    subflowquery = {
+        "size": 500,
+        "query": {"bool": {"should": subflow_query_1 + subflow_query_2}},
+        "_source": ["interactionId"],
+    }
+
+    subs = es.search(body=json.dumps(subflowquery), index="_all")
+
+    sub_interactionIds_match = list(
+        map(
+            lambda res: res["_source"]["interactionId"],
+            subs["hits"]["hits"],
+        )
+    )
+
+    return interactionIds_match + sub_interactionIds_match
+
+
+def find(responseId, field, values, result):
+    # find values in find for all flow for a given responseId
+    # field and result should be one of:
+    #   properties.outcomeDescription.value
+    #   properties.outcomeStatus.value
+    #   properties.outcomeType.value
+    #   properties.outcomeDetail.value
+
+    es = OpenSearch(
+        hosts=[{"host": settings.ES_HOST, "port": settings.ES_PORT}], timeout=300
+    )
+
+    all_interactionIds = get_all_InteractionIds(responseId)
+
+    all_interactionIds_match = list(
+        map(lambda res: {"match": {"interactionId": f"{res}"}}, all_interactionIds)
+    )
+
+    values_match = list(
+        map(
+            lambda res: {"match_phrase": {field: res}},
+            values,
+        )
+    )
+
+    query_found = {
+        "size": 500,
+        "query": {
+            "bool": {
+                "must": [
+                    {
+                        "bool": {
+                            "should": values_match,
+                        }
+                    },
+                    {
+                        "bool": {
+                            "should": all_interactionIds_match,
+                        }
+                    },
+                ]
+            }
+        },
+        "_source": [result],
+    }
+
+    found_result = es.search(body=json.dumps(query_found), index="_all")
+
+    list_found = list(
+        map(
+            lambda x: recursive_decent(x["_source"], result.split(".")),
+            found_result["hits"]["hits"],
+        )
+    )
+
+    return {"found": list_found}
+
+
+def recursive_decent(obj: dict | str, query: list[str]):
+    # given dict and a dot notated key name, return value of key
+    if query == [] or not isinstance(obj, dict):
+        return obj
+    return recursive_decent(obj.get(query[0], ""), query[1:])

requirements.txt

@@ -1,7 +1,7 @@
-fastapi==0.101.1
+fastapi==0.103.2
 uvicorn==0.23.2
 starlette-prometheus==0.9.0
-google-api-python-client==2.97.0
-google-auth-httplib2==0.1.0
-google-auth-oauthlib==1.0.0
-opensearch-py
+google-api-python-client==2.102.0
+google-auth-httplib2==0.1.1
+google-auth-oauthlib==1.1.0
+opensearch-py==2.3.1