Endpoint for uuid query (#1377)

app/__init__.py

@@ -210,6 +210,7 @@ def register_blueprint(application):
     from app.service.callback_rest import service_callback_blueprint
     from app.service.rest import service_blueprint
     from app.status.healthcheck import status as status_blueprint
+    from app.support.rest import support_blueprint
     from app.template.rest import template_blueprint
     from app.template.template_category_rest import template_category_blueprint
     from app.template_folder.rest import template_folder_blueprint
@@ -272,6 +273,8 @@ def register_blueprint(application):
 
     register_notify_blueprint(application, template_category_blueprint, requires_admin_auth)
 
+    register_notify_blueprint(application, support_blueprint, requires_admin_auth, "/support")
+
     register_notify_blueprint(application, cache_blueprint, requires_cache_clear_auth)
 
 

app/support/rest.py

@@ -0,0 +1,129 @@
+from uuid import UUID
+
+from flask import Blueprint, Response, jsonify, request
+from sqlalchemy.orm.exc import NoResultFound
+
+from app.dao.jobs_dao import dao_get_job_by_id
+from app.dao.notifications_dao import get_notification_by_id
+from app.dao.services_dao import dao_fetch_service_by_id
+from app.dao.templates_dao import dao_get_template_by_id
+from app.dao.users_dao import get_user_by_id
+from app.errors import register_errors
+
+support_blueprint = Blueprint("support", __name__)
+register_errors(support_blueprint)
+
+
+def notification_query(id: str) -> dict | None:
+    try:
+        notification = get_notification_by_id(id)
+        if notification:
+            return {
+                "id": notification.id,
+                "type": "notification",
+                "notification_type": notification.notification_type,
+                "status": notification.status,
+                "created_at": notification.created_at,
+                "sent_at": notification.sent_at,
+                "to": notification.to,
+                "service_id": notification.service_id,
+                "service_name": notification.service.name,
+                "template_id": notification.template_id,
+                "template_name": notification.template.name,
+                "job_id": notification.job_id,
+                "job_row_number": notification.job_row_number,
+                "api_key_id": notification.api_key_id,
+            }
+    except NoResultFound:
+        return None
+    return None
+
+
+def template_query(id: str) -> dict | None:
+    try:
+        template = dao_get_template_by_id(id)
+        if template:
+            return {
+                "id": template.id,
+                "type": "template",
+                "template_name": template.name,
+                "service_id": template.service_id,
+                "service_name": template.service.name,
+            }
+    except NoResultFound:
+        return None
+    return None
+
+
+def service_query(id: str) -> dict | None:
+    try:
+        service = dao_fetch_service_by_id(id)
+        if service:
+            return {"id": service.id, "type": "service", "service_name": service.name}
+    except NoResultFound:
+        return None
+    return None
+
+
+def job_query(id: str) -> dict | None:
+    try:
+        job = dao_get_job_by_id(id)
+        if job:
+            return {
+                "id": job.id,
+                "type": "job",
+                "original_file_name": job.original_file_name,
+                "created_at": job.created_at,
+                "created_by_id": job.created_by_id,
+                "created_by_name": job.created_by.name,
+                "processing_started": job.processing_started,
+                "processing_finished": job.processing_finished,
+                "notification_count": job.notification_count,
+                "job_status": job.job_status,
+                "service_id": job.service_id,
+                "service_name": job.service.name,
+                "template_id": job.template_id,
+                "template_name": job.template.name,
+            }
+    except NoResultFound:
+        return None
+    return None
+
+
+def user_query(id: str) -> dict | None:
+    try:
+        user = get_user_by_id(id)
+        if user:
+            return {
+                "id": user.id,
+                "type": "user",
+                "user_name": user.name,
+            }
+    except NoResultFound:
+        return None
+    return None
+
+
+@support_blueprint.route("/find-ids", methods=["GET"])
+def find_ids() -> tuple[Response, int]:
+    ids = request.args.get("ids")
+    if not ids:
+        return jsonify({"error": "no ids provided"}), 400
+
+    info = []
+    for id in ids.replace(",", " ").split():
+        try:
+            UUID(id)
+        except ValueError:
+            info.append({"id": id, "type": "not a uuid"})
+            continue
+        query_funcs = [user_query, service_query, template_query, job_query, notification_query]
+        results = None
+        for query_func in query_funcs:
+            results = query_func(id)
+            if results:
+                info.append(results)
+                break
+        if not results:
+            info.append({"id": id, "type": "no result found"})
+    return jsonify(info), 200

tests/app/support/test_rest.py

@@ -0,0 +1,87 @@
+import uuid
+
+import pytest
+
+
+def test_find_ids_user(admin_request, sample_user):
+    json_resp = admin_request.get("support.find_ids", ids=sample_user.id)[0]
+    assert json_resp["type"] == "user"
+    assert json_resp["id"] == str(sample_user.id)
+    assert json_resp["user_name"] == sample_user.name
+
+
+def test_find_ids_service(admin_request, sample_service):
+    json_resp = admin_request.get("support.find_ids", ids=sample_service.id)[0]
+    assert json_resp["type"] == "service"
+    assert json_resp["id"] == str(sample_service.id)
+    assert json_resp["service_name"] == sample_service.name
+
+
+def test_find_ids_template(admin_request, sample_template):
+    json_resp = admin_request.get("support.find_ids", ids=sample_template.id)[0]
+    assert json_resp["type"] == "template"
+    assert json_resp["id"] == str(sample_template.id)
+    assert json_resp["template_name"] == sample_template.name
+    assert json_resp["service_id"] == str(sample_template.service_id)
+    assert json_resp["service_name"] == sample_template.service.name
+
+
+def test_find_ids_job(admin_request, sample_job):
+    json_resp = admin_request.get("support.find_ids", ids=sample_job.id)[0]
+    assert json_resp["type"] == "job"
+    assert json_resp["id"] == str(sample_job.id)
+    assert json_resp["original_file_name"] == sample_job.original_file_name
+    assert json_resp["created_by_id"] == str(sample_job.created_by_id)
+    assert json_resp["created_by_name"] == sample_job.created_by.name
+    assert json_resp["notification_count"] == sample_job.notification_count
+    assert json_resp["job_status"] == sample_job.job_status
+    assert json_resp["service_id"] == str(sample_job.service_id)
+    assert json_resp["service_name"] == sample_job.service.name
+    assert json_resp["template_id"] == str(sample_job.template_id)
+    assert json_resp["template_name"] == sample_job.template.name
+
+
+def test_find_ids_notification(admin_request, sample_notification_with_job):
+    json_resp = admin_request.get("support.find_ids", ids=sample_notification_with_job.id)[0]
+    assert json_resp["type"] == "notification"
+    assert json_resp["id"] == str(sample_notification_with_job.id)
+    assert json_resp["notification_type"] == sample_notification_with_job.notification_type
+    assert json_resp["status"] == sample_notification_with_job.status
+    assert json_resp["to"] == sample_notification_with_job.to
+    assert json_resp["service_id"] == str(sample_notification_with_job.service_id)
+    assert json_resp["service_name"] == sample_notification_with_job.service.name
+    assert json_resp["template_id"] == str(sample_notification_with_job.template_id)
+    assert json_resp["template_name"] == sample_notification_with_job.template.name
+    assert json_resp["job_id"] == str(sample_notification_with_job.job_id)
+    assert json_resp["job_row_number"] == sample_notification_with_job.job_row_number
+    assert json_resp["api_key_id"] is None
+
+
+def test_find_ids_unknown_uuid(admin_request, sample_user):
+    search_uuid = str(uuid.uuid4())
+    json_resp = admin_request.get("support.find_ids", ids=search_uuid)[0]
+    assert json_resp["type"] == "no result found"
+
+
+def test_find_ids_no_ids(admin_request):
+    json_resp = admin_request.get("support.find_ids", _expected_status=400, ids=None)
+    assert json_resp == {"error": "no ids provided"}
+
+
+def test_find_ids_empty_ids(admin_request):
+    json_resp = admin_request.get("support.find_ids", _expected_status=400, ids=[])
+    assert json_resp == {"error": "no ids provided"}
+
+
+def test_find_ids_id_not_uuid(admin_request):
+    search_uuid = "hello"
+    json_resp = admin_request.get("support.find_ids", ids=search_uuid)[0]
+    assert json_resp["type"] == "not a uuid"
+
+
+@pytest.mark.parametrize("delimiter", [",", " ", "  ,\n\n, "])
+def test_find_ids_two_ids(admin_request, sample_user, sample_service, delimiter):
+    json_resp = admin_request.get("support.find_ids", ids=f"{sample_user.id}{delimiter}{sample_service.id}")
+    assert len(json_resp) == 2
+    assert json_resp[0]["type"] == "user"
+    assert json_resp[1]["type"] == "service"