Adding an email validation route to the qualtrix service

qualtrix/api.py

@@ -5,7 +5,7 @@
 import logging
 
 import fastapi
-from fastapi import HTTPException
+from fastapi import HTTPException, responses
 from pydantic import BaseModel, typing
 
 from qualtrix import client, error
@@ -27,6 +27,11 @@ class SessionModel(SurveyModel):
     sessionId: str
 
 
+class EmailValidationModel(BaseModel):
+    directoryId: str
+    email: str
+
+
 @router.post("/bulk-responses")
 async def test():
     return client.result_export()
@@ -54,3 +59,21 @@ async def session(request: SessionModel):
         return client.delete_session(request.surveyId, request.sessionId)
     except error.QualtricsError as e:
         raise HTTPException(status_code=400, detail=e.args)
+
+
+@router.post("/email")
+async def email(request: EmailValidationModel):
+    """
+    Checks if the supplied email exists in the Contact Pool
+    """
+    try:
+        return responses.JSONResponse(
+            status_code=200,
+            content={
+                "emailPresent": client.validate_email(
+                    request.email, request.directoryId
+                )
+            },
+        )
+    except error.QualtricsError as e:
+        raise HTTPException(status_code=422, detail=e.args)

qualtrix/client.py

@@ -11,6 +11,10 @@
 # Permisions # read:survey_responses
 
 auth_header = {"X-API-TOKEN": settings.API_TOKEN}
+application_header = {
+    "X-API-TOKEN": settings.API_TOKEN,
+    "content-type": "application/json",
+}
 
 
 def get_response(survey_id: str, response_id: str):
@@ -67,6 +71,21 @@ def get_survey_schema(survey_id: str):
     return r.json()
 
 
+def validate_email(email: str, directory_id=settings.DIRECTORY_ID):
+    r = requests.post(
+        settings.BASE_URL
+        + f"/directories/{directory_id}/contacts/search?includeEmbedded=true&includeSegments=false",
+        headers=application_header,
+        json=({"filter": {"filterType": "email", "comparison": "eq", "value": email}}),
+    )
+
+    if r.status_code != 200:
+        raise error.QualtricsError("Survey response not found, likely bad POOL ID")
+
+    elements = list((dict(r.json())["result"]["elements"]))
+    return len(elements) != 0
+
+
 def result_export(survey_id: str):
     r_body = {
         "format": "json",