wSubmissions (#45)

submissions: authors initial implementation

Co-authored-by: DonHaul <[email protected]>

.envs/docker/.django

@@ -17,3 +17,7 @@ CELERY_FLOWER_PASSWORD=debug
 # OpenSearch
 OPENSEARCH_HOST=opensearch:9200
 OPENSEARCH_INDEX_PREFIX=backoffice-backend-local
+
+# Airflow
+AIRFLOW_BASE_URL=http://localhost:8080
+AIRFLOW_TOKEN=CHANGE_ME

.envs/local/.django

@@ -17,3 +17,7 @@ CELERY_FLOWER_PASSWORD=debug
 # Opensearch
 OPENSEARCH_HOST=opensearch:9200
 OPENSEARCH_INDEX_PREFIX=backoffice-backend-local
+
+# Airflow
+AIRFLOW_BASE_URL=http://host.docker.internal:8082
+AIRFLOW_TOKEN=CHANGE_ME

README.md

@@ -11,6 +11,15 @@ License: MIT
 
 Moved to [settings](http://cookiecutter-django.readthedocs.io/en/latest/settings.html).
 
+## Quickstart
+
+1. `docker compose -f local.yml run -d`
+2. Enter django container backoffice-local-django and execute the following commands
+   a. `python manage.py create_groups` to create author and curator group definitions in the db
+   b. `python manage.py createsuperuser` to create a super user
+3. Navigate to http://localhost:8000/admin/authtoken/ login with the newly created user and assign a token to it
+4. Set your user to be in the admin group in here http://localhost:8000/admin/users/user/
+
 ## Basic Commands
 
 ### Setting Up Your Users

backoffice/workflows/airflow_utils.py

@@ -0,0 +1,34 @@
+from os import environ
+
+import requests
+from django.http import JsonResponse
+from requests.exceptions import RequestException
+from rest_framework import status
+
+AIRFLOW_BASE_URL = environ.get("AIRFLOW_BASE_URL")
+
+AIRFLOW_HEADERS = {"Content-Type": "application/json", "Authorization": f"Basic {environ.get('AIRFLOW_TOKEN')}"}
+
+
+def trigger_airflow_dag(dag_id, workflow_id, extra_data=None):
+    """Triggers an airflow dag.
+
+    :param dag_id: name of the dag to run
+    :param workflow_id: id of the workflow being triggered
+    :returns: request response
+    """
+
+    data = {"dag_run_id": workflow_id, "conf": {"workflow_id": workflow_id}}
+
+    if extra_data is not None:
+        data["conf"].update(extra_data)
+
+    url = f"{AIRFLOW_BASE_URL}/api/v1/dags/{dag_id}/dagRuns"
+
+    try:
+        response = requests.post(url, json=data, headers=AIRFLOW_HEADERS)
+        response.raise_for_status()
+        return JsonResponse(response.json())
+    except RequestException as req_err:
+        data = {"error": req_err}
+        return JsonResponse(data, status=status.HTTP_500_INTERNAL_SERVER_ERROR)

backoffice/workflows/api/serializers.py

@@ -4,6 +4,8 @@
 from backoffice.workflows.documents import WorkflowDocument
 from backoffice.workflows.models import Workflow, WorkflowTicket
 
+from ..constants import ResolutionDags
+
 
 class WorkflowSerializer(serializers.ModelSerializer):
     class Meta:
@@ -21,3 +23,8 @@ class WorkflowDocumentSerializer(DocumentSerializer):
     class Meta:
         document = WorkflowDocument
         fields = "__all__"
+
+
+class AuthorResolutionSerializer(serializers.Serializer):
+    value = serializers.ChoiceField(choices=ResolutionDags)
+    create_ticket = serializers.BooleanField(default=False)

backoffice/workflows/api/views.py

@@ -1,13 +1,21 @@
 from django.shortcuts import get_object_or_404
 from django_elasticsearch_dsl_drf.viewsets import BaseDocumentViewSet
 from rest_framework import status, viewsets
+from rest_framework.decorators import action
 from rest_framework.response import Response
 
 from backoffice.utils.pagination import OSStandardResultsSetPagination
+from backoffice.workflows import airflow_utils
 from backoffice.workflows.documents import WorkflowDocument
 from backoffice.workflows.models import Workflow, WorkflowTicket
 
-from .serializers import WorkflowDocumentSerializer, WorkflowSerializer, WorkflowTicketSerializer
+from ..constants import WORKFLOW_DAG, ResolutionDags
+from .serializers import (
+    AuthorResolutionSerializer,
+    WorkflowDocumentSerializer,
+    WorkflowSerializer,
+    WorkflowTicketSerializer,
+)
 
 
 class WorkflowViewSet(viewsets.ModelViewSet):
@@ -70,6 +78,27 @@ def create(self, request, *args, **kwargs):
             return Response({"error": str(e)}, status=status.HTTP_500_INTERNAL_SERVER_ERROR)
 
 
+class AuthorWorkflowViewSet(viewsets.ViewSet):
+    serializer_class = WorkflowSerializer
+
+    def create(self, request):
+        serializer = self.serializer_class(data=request.data)
+        if serializer.is_valid(raise_exception=True):
+            workflow = Workflow.objects.create(
+                data=serializer.validated_data["data"], workflow_type=serializer.validated_data["workflow_type"]
+            )
+        return airflow_utils.trigger_airflow_dag(WORKFLOW_DAG[workflow.workflow_type], str(workflow.id), workflow.data)
+
+    @action(detail=True, methods=["post"])
+    def resolve(self, request, pk=None):
+        serializer = AuthorResolutionSerializer(data=request.data)
+        if serializer.is_valid(raise_exception=True):
+            extra_data = {"create_ticket": serializer.validated_data["create_ticket"]}
+            return airflow_utils.trigger_airflow_dag(
+                ResolutionDags[serializer.validated_data["value"]].label, pk, extra_data
+            )
+
+
 class WorkflowDocumentView(BaseDocumentViewSet):
     def __init__(self, *args, **kwargs):
         super().__init__(*args, **kwargs)

backoffice/workflows/constants.py

@@ -1,3 +1,5 @@
+from django.db import models
+
 # tickets
 TICKET_TYPES = (
     ("author_create_curation", "Author create curation"),
@@ -6,18 +8,36 @@
 DEFAULT_TICKET_TYPE = "author_create_curation"
 
 # workflows
-DEFAULT_STATUS_CHOICE = "running"
-DEFAULT_WORKFLOW_TYPE = "HEP_create"
-STATUS_CHOICES = (
-    ("running", "Running"),
-    ("approval", "Waiting for approval"),
-    ("completed", "Completed"),
-    ("error", "Error"),
-)
 
-WORKFLOW_TYPES = (
-    ("HEP_CREATE", "HEP create"),
-    ("HEP_UPDATE", "HEP update"),
-    ("AUTHOR_CREATE", "Author create"),
-    ("AUTHOR_UPDATE", "Author update"),
-)
+
+class StatusChoices(models.TextChoices):
+    RUNNING = "running", "Running"
+    APPROVAL = "approval", "Waiting for approva"
+    COMPLETED = "completed", "Completed"
+    ERROR = "error", "Error"
+
+
+DEFAULT_STATUS_CHOICE = StatusChoices.RUNNING
+
+
+class WorkflowType(models.TextChoices):
+    HEP_CREATE = "HEP_CREATE", "HEP create"
+    HEP_UPDATE = "HEP_UPDATE", "HEP update"
+    AUTHOR_CREATE = "AUTHOR_CREATE", "Author create"
+    AUTHOR_UPDATE = "AUTHOR_UPDATE", "Author update"
+
+
+DEFAULT_WORKFLOW_TYPE = WorkflowType.HEP_CREATE
+
+# author dags for each workflow type
+WORKFLOW_DAG = {
+    WorkflowType.HEP_CREATE: "",
+    WorkflowType.HEP_UPDATE: "",
+    WorkflowType.AUTHOR_CREATE: "author_create_initialization_dag",
+    WorkflowType.AUTHOR_UPDATE: "author_update_dag",
+}
+
+
+class ResolutionDags(models.TextChoices):
+    accept = "accept", "author_create_approved_dag"
+    reject = "reject", "author_create_rejected_dag"

backoffice/workflows/migrations/0008_alter_workflow_status_alter_workflow_workflow_type.py

@@ -0,0 +1,40 @@
+# Generated by Django 4.2.6 on 2024-07-09 12:42
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ("workflows", "0007_alter_workflow_core_alter_workflow_is_update"),
+    ]
+
+    operations = [
+        migrations.AlterField(
+            model_name="workflow",
+            name="status",
+            field=models.CharField(
+                choices=[
+                    ("running", "Running"),
+                    ("approval", "Waiting for approva"),
+                    ("completed", "Completed"),
+                    ("error", "Error"),
+                ],
+                default="running",
+                max_length=30,
+            ),
+        ),
+        migrations.AlterField(
+            model_name="workflow",
+            name="workflow_type",
+            field=models.CharField(
+                choices=[
+                    ("HEP_CREATE", "HEP create"),
+                    ("HEP_UPDATE", "HEP update"),
+                    ("AUTHOR_CREATE", "Author create"),
+                    ("AUTHOR_UPDATE", "Author update"),
+                ],
+                default="HEP_CREATE",
+                max_length=30,
+            ),
+        ),
+    ]

backoffice/workflows/models.py

@@ -6,9 +6,9 @@
     DEFAULT_STATUS_CHOICE,
     DEFAULT_TICKET_TYPE,
     DEFAULT_WORKFLOW_TYPE,
-    STATUS_CHOICES,
     TICKET_TYPES,
-    WORKFLOW_TYPES,
+    StatusChoices,
+    WorkflowType,
 )
 
 
@@ -17,13 +17,13 @@ class Workflow(models.Model):
 
     workflow_type = models.CharField(
         max_length=30,
-        choices=WORKFLOW_TYPES,
+        choices=WorkflowType.choices,
         default=DEFAULT_WORKFLOW_TYPE,
     )
     data = models.JSONField()
     status = models.CharField(
         max_length=30,
-        choices=STATUS_CHOICES,
+        choices=StatusChoices.choices,
         default=DEFAULT_STATUS_CHOICE,
     )
     core = models.BooleanField(default=False)

backoffice/workflows/tests/test_views.py

@@ -1,11 +1,16 @@
+from unittest.mock import patch
+
 from django.apps import apps
 from django.contrib.auth import get_user_model
 from django.contrib.auth.models import Group
 from django.test import TransactionTestCase
+from django.urls import reverse
 from opensearch_dsl import Index
+from rest_framework import status
 from rest_framework.test import APIClient
 
 from backoffice.workflows.api.serializers import WorkflowTicketSerializer
+from backoffice.workflows.constants import StatusChoices
 from backoffice.workflows.models import WorkflowTicket
 
 User = get_user_model()
@@ -37,7 +42,7 @@ class TestWorkflowViewSet(BaseTransactionTestCase):
 
     def setUp(self):
         super().setUp()
-        self.workflow = Workflow.objects.create(data={}, status="approval", core=True, is_update=False)
+        self.workflow = Workflow.objects.create(data={}, status=StatusChoices.APPROVAL, core=True, is_update=False)
 
     def test_list_curator(self):
         self.api_client.force_authenticate(user=self.curator)
@@ -70,7 +75,7 @@ def setUp(self):
         super().setUp()
         index = Index("backoffice-backend-test-workflows")
         index.delete(ignore=[400, 404])
-        self.workflow = Workflow.objects.create(data={}, status="approval", core=True, is_update=False)
+        self.workflow = Workflow.objects.create(data={}, status=StatusChoices.APPROVAL, core=True, is_update=False)
 
     def test_list_curator(self):
         self.api_client.force_authenticate(user=self.curator)
@@ -100,7 +105,7 @@ class TestWorkflowPartialUpdateViewSet(BaseTransactionTestCase):
 
     def setUp(self):
         super().setUp()
-        self.workflow = Workflow.objects.create(data={}, status="approval", core=True, is_update=False)
+        self.workflow = Workflow.objects.create(data={}, status=StatusChoices.APPROVAL, core=True, is_update=False)
 
     @property
     def endpoint(self):
@@ -198,3 +203,66 @@ def test_create_happy_flow(self):
         assert "ticket_type" in response.data
 
         assert response.data == WorkflowTicketSerializer(WorkflowTicket.objects.last()).data
+
+
+class TestAuthorWorkflowViewSet(BaseTransactionTestCase):
+    endpoint = "/api/authors/"
+    reset_sequences = True
+    fixtures = ["backoffice/fixtures/groups.json"]
+
+    @patch("backoffice.workflows.airflow_utils.requests.post")
+    def test_create_author(self, mock_post):
+        self.api_client.force_authenticate(user=self.curator)
+
+        mock_response = mock_post.return_value
+        mock_response.status_code = status.HTTP_200_OK
+        mock_response.json.return_value = {"key": "value"}
+
+        data = {
+            "workflow_type": "AUTHOR_CREATE",
+            "status": "running",
+            "data": {
+                "native_name": "NATIVE_NAME",
+                "alternate_name": "NAME",
+                "display_name": "FIRST_NAME",
+                "family_name": "LAST_NAME",
+                "given_name": "GIVEN_NAME",
+            },
+        }
+
+        url = reverse("api:workflows-authors-list")
+        response = self.api_client.post(url, format="json", data=data)
+
+        self.assertEqual(response.status_code, 200)
+
+    @patch("backoffice.workflows.airflow_utils.requests.post")
+    def test_accept_author(self, mock_post):
+        self.api_client.force_authenticate(user=self.curator)
+
+        mock_response = mock_post.return_value
+        mock_response.status_code = status.HTTP_200_OK
+        mock_response.json.return_value = {"key": "value"}
+
+        data = {"create_ticket": True, "value": "accept"}
+
+        response = self.api_client.post(
+            reverse("api:workflows-authors-resolve", kwargs={"pk": "WORKFLOW_ID"}), format="json", data=data
+        )
+
+        self.assertEqual(response.status_code, 200)
+
+    @patch("backoffice.workflows.airflow_utils.requests.post")
+    def test_reject_author(self, mock_post):
+        self.api_client.force_authenticate(user=self.curator)
+
+        mock_response = mock_post.return_value
+        mock_response.status_code = status.HTTP_200_OK
+        mock_response.json.return_value = {"key": "value"}
+
+        data = {"create_ticket": True, "value": "reject"}
+
+        response = self.api_client.post(
+            reverse("api:workflows-authors-resolve", kwargs={"pk": "WORKFLOW_ID"}), format="json", data=data
+        )
+
+        self.assertEqual(response.status_code, 200)