flows: remove invenio webhooks

Author: kpsherva

cds/modules/deposit/api.py

@@ -63,7 +63,7 @@
 from ..records.tasks import create_symlinks
 from ..records.utils import lowercase_value
 from ..records.validators import PartialDraft4Validator
-from ..webhooks.status import (get_deposit_events, get_tasks_status_by_task,
+from ..webhooks.status import (get_deposit_flows, get_tasks_status_by_task,
                                merge_tasks_status)
 from .errors import DiscardConflict
 from .resolver import get_video_pid
@@ -931,7 +931,7 @@ def edit(self, pid=None):
 
     def _clean_tasks(self):
         """Clean all tasks."""
-        events = get_deposit_events(deposit_id=self['_deposit']['id'])
+        events = get_deposit_flows(deposit_id=self['_deposit']['id'])
         for event in events:
             event.receiver.delete(event=event)
 
@@ -1004,7 +1004,7 @@ def get_report_number_sequence(self, **kwargs):
     def _current_tasks_status(self):
         """Return up-to-date tasks status."""
         return get_tasks_status_by_task(
-            get_deposit_events(self['_deposit']['id']),
+            get_deposit_flows(self['_deposit']['id']),
             statuses=deepcopy(self['_cds'].get('state', {})))
 
     def generate_duration(self):

cds/modules/deposit/static/js/cds_deposit/avc/avc.module.js

@@ -102,9 +102,9 @@ function cdsDepositsConfig(
     iiif: '/api/iiif/v2/<%=deposit%>:<%=version_id%>:<%=key%>/full/!<%=res%>/0/default.png',
     categories: '/api/categories',
     video: '/deposit/<%=deposit%>/preview/video/<%=key%>',
-    eventInfo: '/hooks/receivers/avc/events/<%=eventId%>',
-    restartEvent: '/hooks/receivers/avc/events/<%=eventId%>/tasks/<%=taskId%>',
-    taskFeedback: '/hooks/receivers/avc/events/<%=eventId%>/feedback',
+    eventInfo: '/hooks/receivers/avc/flows/<%=eventId%>',
+    restartEvent: '/hooks/receivers/avc/flows/<%=eventId%>/tasks/<%=taskId%>',
+    taskFeedback: '/hooks/receivers/avc/flows/<%=eventId%>/feedback',
     selfVideo: '/api/deposits/video/<%=deposit%>',
     bucketVideo: '/api/files/<%=bucket%>',
     actionVideo: '/api/deposits/video/<%=deposit%>/actions/<%=action%>',

cds/modules/deposit/static/js/cds_deposit/avc/components/cdsDeposit.js

@@ -209,7 +209,7 @@ function cdsDepositCtrl(
 
     this.restartFailedSubformats = function(subformatKeys) {
       var master = that.findMasterFile();
-      var eventId = master.tags._event_id;
+      var eventId = master.tags._flow_id;
       master.subformat.forEach(
         function(subformat) {
           if (subformatKeys.includes(subformat.key)) {
@@ -223,7 +223,7 @@ function cdsDepositCtrl(
           var restartEvents = data.filter(function(taskInfo) {
             return subformatKeys.includes(taskInfo.info.payload.key);
           }).map(function(taskInfo) {
-            var eventId = taskInfo.info.payload.event_id;
+            var eventId = taskInfo.info.payload.flow_id;
             var taskId = taskInfo.id;
             return that.restartEvent(eventId, taskId);
           });
@@ -275,7 +275,7 @@ function cdsDepositCtrl(
 
         var fetchPresetsPromise = $q.resolve();
         if (that.presets && that.presets.length == 0) {
-          var eventId = masterFile.tags._event_id;
+          var eventId = masterFile.tags._flow_id;
           if (eventId) {
             var eventUrl = urlBuilder.eventInfo({eventId: eventId});
             var updatePresets = function (resp) {
@@ -344,7 +344,7 @@ function cdsDepositCtrl(
       // Update only if it is ``draft``
       if (that.isDraft()){
         var masterFile = that.findMasterFile();
-        var eventId = _.get(masterFile, 'tags._event_id', undefined);
+        var eventId = _.get(masterFile, 'tags._flow_id', undefined);
         that.webhookEventId = eventId;
         if (eventId) {
           that.getTaskFeedback(eventId)

cds/modules/deposit/static/js/cds_deposit/avc/components/cdsUploader.js

@@ -347,7 +347,7 @@ function cdsUploaderCtrl(
             that.files.push(newMasterFile);
             that.queue.push(newMasterFile);
             // Upload the video file
-            var old_event_id = old_master[0]['tags']['_event_id']
+            var old_event_id = old_master[0]['tags']['_flow_id']
             that.deleteEvent(old_event_id).then(
               function success(response) {
                 that.cdsDepositCtrl.previewer = null;

cds/modules/deposit/static/templates/cds_deposit/deposits.html

@@ -100,8 +100,8 @@ <h3>Click here to select videos to upload</h3>
                   dropbox-enabled="true"
                   dropbox-selector=".dropbox-upload"
                   dropbox-app-key="{{ $ctrl.dropboxAppKey }}"
-                  remote-master-receiver="/api/hooks/receivers/avc/events/"
-                  remote-children-receiver="/api/hooks/receivers/downloader/events/"
+                  remote-master-receiver="/api/hooks/receivers/avc/flows/"
+                  remote-children-receiver="/api/hooks/receivers/downloader/flows/"
               >
                 <div class="dropbox-upload"></div>
               </cds-remote-uploader>
@@ -138,15 +138,15 @@ <h3>Click here to select videos to upload</h3>
                     auto-start-upload="true"
                     files="child._files"
                     template="/static/templates/cds_deposit/types/video/uploader.html"
-                    remote-master-receiver="/api/hooks/receivers/avc/events/"
+                    remote-master-receiver="/api/hooks/receivers/avc/flows/"
                   >
                     <cds-remote-uploader
                         template="/static/templates/cds_deposit/remote_upload.html"
                         dropbox-enabled="true"
                         dropbox-selector=".dropbox-upload"
                         dropbox-app-key="{{ $ctrl.dropboxAppKey }}"
-                        remote-master-receiver="/api/hooks/receivers/avc/events/"
-                        remote-children-receiver="/api/hooks/receivers/downloader/events/"
+                        remote-master-receiver="/api/hooks/receivers/avc/flows/"
+                        remote-children-receiver="/api/hooks/receivers/downloader/flows/"
                     >
                       <div class="dropbox-upload"></div>
                     </cds-remote-uploader>

cds/modules/flows/api.py

@@ -24,7 +24,7 @@
 # as an Intergovernmental Organization or submit itself to any jurisdiction.
 
 """Invenio-Flow python API."""
-
+import json
 import logging
 from functools import wraps
 from itertools import repeat
@@ -94,18 +94,37 @@ def payload(self):
         """Get flow payload."""
         return self.model.payload if self.model else None
 
-    @property
-    def previous_id(self):
-        """Get the previous run of the flow."""
-        return self.model.previous_id if self.model else None
-
     @payload.setter
     def payload(self, value):
         """Update payload."""
         if self.model:
             self.model.payload = value
             db.session.merge(self.model)
 
+    @property
+    def response(self):
+        """Get flow payload."""
+        return self.model.response if self.model else None
+
+    @response.setter
+    def response(self, value):
+        """Update payload."""
+        if self.model:
+            self.model.response = value
+            db.session.merge(self.model)
+
+    @property
+    def response_code(self):
+        """Get flow payload."""
+        return self.model.response_code if self.model else None
+
+    @response_code.setter
+    def response_code(self, value):
+        """Update payload."""
+        if self.model:
+            self.model.response_code = value
+            db.session.merge(self.model)
+
     @property
     def created(self):
         """Get creation timestamp."""
@@ -117,30 +136,40 @@ def updated(self):
         return self.model.updated if self.model else None
 
     @property
-    def status(self):
+    def json(self):
         """Get flow status."""
         if self.model is None:
             return None
 
         res = self.model.to_dict()
-        res.update({'tasks': [t.to_dict() for t in self.model.tasks]})
+        res.update(
+            {'tasks': [t.to_dict() for t in self.model.tasks]}
+        )
         return res
 
+    @property
+    def status(self):
+        return self.model.status if self.model else None
+
     @classmethod
     def get_flow(cls, id_):
         """Retrieve a Flow from the database by Id."""
         obj = FlowModel.get(id_)
         return cls(obj)
 
     @classmethod
-    def create(cls, name, payload=None, id_=None, previous_id=None):
+    def create(cls, name, payload=None,
+               id_=None, user_id=None,
+               deposit_id=None, receiver_id=None):
         """Create a new flow instance and store it in the database.."""
         with db.session.begin_nested():
             obj = FlowModel(
                 name=name,
                 id=id_ or uuid(),
                 payload=payload or dict(),
-                previous_id=previous_id,
+                user_id=user_id,
+                deposit_id=deposit_id,
+                receiver_id=receiver_id
             )
             db.session.add(obj)
         logger.info('Created new Flow %s', obj)
@@ -163,11 +192,11 @@ def _new_task(self, task, kwargs, previous):
         task_id = uuid()
         kwargs = kwargs if kwargs else {}
         kwargs.update({'flow_id': str(self.id), 'task_id': task_id})
-        kwargs.update(self.payload,)  # TODO: Do we need to move this to a key?
+        kwargs.update(self.payload, )
         signature = task.subtask(
             task_id=task_id,
             kwargs=kwargs,
-            immutable=True,  # TODO, ad this as an option/parameter?
+            immutable=True,
         )
 
         _ = TaskModel.create(
@@ -181,7 +210,7 @@ def _new_task(self, task, kwargs, previous):
         return signature
 
     def build(self):
-        """."""
+        """Build flow."""
         raise NotImplementedError()
 
     def assemble(self, build_func):
@@ -221,7 +250,7 @@ def assemble(self, build_func):
     def start(self):
         """Start the flow asynchronously."""
         if not self._canvas:
-            self.assemble()
+            self.assemble(self.build)
         return self._canvas.apply_async()
 
     def stop(self):
@@ -266,15 +295,13 @@ def restart_task(self, task_id):
 
         kwargs = {'flow_id': str(self.id), 'task_id': str(task.id)}
         kwargs.update(task.payload)
-        kwargs.update(self.payload)  # TODO: Do we need to move this to a key?
+        kwargs.update(self.payload)
         return (
-            celery_app.tasks.get(task.name)
-            .subtask(
-                task_id=str(task.id), 
+            celery_app.tasks.get(task.name).subtask(
+                task_id=str(task.id),
                 kwargs=kwargs,
-                immutable=True,  # TODO, ad this as an option/parameter?
-            )
-            .apply_async()
+                immutable=True,
+            ).apply_async()
         )
 
 
@@ -308,3 +335,42 @@ def on_success(self, retval, task_id, args, kwargs):
             '{}'.format(retval),
         )
         super(Task, self).on_success(retval, task_id, args, kwargs)
+
+    @staticmethod
+    def build_task_json_status(task_json):
+        """."""
+        from ..webhooks.status import TASK_NAMES
+        # Get the UI name of the task
+        task_name = TASK_NAMES.get(task_json['name'])
+
+        # Add the information the UI needs on the right position
+        payload = task_json['payload']
+        payload['type'] = task_name
+
+        payload['key'] = payload.get('preset_quality', payload['key'])
+
+        if task_name == 'file_video_metadata_extraction':
+            # try to load message as JSON,
+            # we only need this for this particular task
+            try:
+                payload['extracted_metadata'] = \
+                    json.loads(task_json['message'])
+            except ValueError:
+                payload['extracted_metadata'] = task_json['message']
+
+            task_json['message'] = 'Attached video metadata'
+
+        celery_task_status = 'REVOKED' if 'Not transcoding' in task_json[
+            'message'] else task_json['status']
+
+        task_status = {
+            'name': task_name,
+            'id': task_json['id'],
+            'status': celery_task_status,
+            'info': {
+                'payload': payload,
+                'message': task_json['message'],
+            },
+        }
+
+        return task_status

cds/modules/flows/migration.py

@@ -31,15 +31,18 @@
 from ..deposit.api import deposit_video_resolver
 from ..records.api import CDSVideosFilesIterator
 from .models import Status
+from ..webhooks.receivers import AVCWorkflow
 
 
-def migrate_event(event):
+def migrate_event(deposit_id):
     """Migrate an old event into Flows."""
-    receiver = event.receiver
-    flow = receiver._workflow(event=event)
+    receiver = AVCWorkflow()
+    deposit = deposit_video_resolver(deposit_id)
+    import ipdb;ipdb.set_trace()
+
+    flow = receiver._workflow(deposit_id)
 
     # Update flow task status depending on the content of th record
-    deposit_id = event.payload['deposit_id']
     deposit = deposit_video_resolver(deposit_id)
 
     original_file = CDSVideosFilesIterator.get_master_video_file(deposit)