Add basic dramatiq implementation

TODO: Add pipeline stages for updating DB after execution

datalab/datalab_session/serializers.py

@@ -1,10 +1,12 @@
 from rest_framework import serializers
 
 from datalab.datalab_session.models import DataSession, DataOperation
+from datalab.datalab_session.data_operations.data_operation import available_operations
 
 
 class DataOperationSerializer(serializers.ModelSerializer):
     session_id = serializers.IntegerField(write_only=True, required=False)
+    name = serializers.ChoiceField(choices=[name for name in available_operations().keys()])
 
     class Meta:
         model = DataOperation

datalab/datalab_session/viewsets.py

@@ -5,7 +5,7 @@
 from datalab.datalab_session.serializers import DataSessionSerializer, DataOperationSerializer
 from datalab.datalab_session.models import DataSession, DataOperation
 from datalab.datalab_session.filters import DataSessionFilterSet
-
+from datalab.datalab_session.tasks import execute_data_operation
 
 class DataOperationViewSet(viewsets.ModelViewSet):
     serializer_class = DataOperationSerializer
@@ -14,6 +14,7 @@ def get_queryset(self):
         return DataOperation.objects.filter(session=self.kwargs['session_pk'])
 
     def perform_create(self, serializer):
+        execute_data_operation.send(serializer.validated_data['name'], serializer.validated_data['input_data'])
         serializer.save(session_id=self.kwargs['session_pk'])
 
 

pyproject.toml

@@ -16,6 +16,9 @@ drf-nested-routers = "^0.93.4"
 ocs-authentication = "^0.2.2"
 psycopg2-binary = "^2.9.9"
 django-dramatiq = "^0.11.6"
+redis = "^5.0.1"
+rabbitmq = "^0.2.0"
+pika = "^1.3.2"
 
 [tool.poetry.group.dev.dependencies]
 pytest = "^7.4.3"