Small concurrency and logging fix

ayta/blueprints/admin.py

@@ -209,12 +209,9 @@ def queue():
             get_nosql().queue_emptyQueue()
             flash(f'Queue has been emptied')
             
-        elif task == 'run-download':
-            get_nosql().queue_emptyQueue()
-            flash(f'Queue has been emptied')
-            
         elif task == 'queue-run-once':
-            video_queue.delay()
+            task =  video_queue.delay()
+            flash(f'Task has been started on the oldest queued item: {task.id}')
         
         return redirect(url_for('admin.queue'))
         

ayta/nosql.py

@@ -411,13 +411,17 @@ class Mango:
     ##########################################
         
     def queue_insertQueue(self, videoId, endpointId=None):
-        # if no document exists
+        # if already queued
-        if not self.download_queue.count_documents({'id': videoId}) >= 1:
+        if self.download_queue.count_documents({'id': videoId}) >= 1:
-            self.download_queue.insert_one({'id': videoId, 'endpoint': endpointId, 'created_time': current_time(object=True), 'status': 'queued'}).inserted_id
+            return False
-            return True
         
-        # key already in queue
+        # if already in archive
-        return False
+        if self.check_exists(videoId):
+            return False
+        
+        # add to queue
+        self.download_queue.insert_one({'id': videoId, 'endpoint': endpointId, 'created_time': current_time(object=True), 'status': 'queued'}).inserted_id
+        return True
         
     def queue_deleteQueue(self, videoId):
         if self.download_queue.delete_one({'id': videoId}):
@@ -430,18 +434,20 @@ class Mango:
     def queue_emptyQueue(self):
         return self.download_queue.delete_many({})
         
+    def queue_setFailed(self, videoId):
+        return self.download_queue.update_one({'id': videoId}, {'$set': {'status': 'failed'}})
+        
     def queue_getNext(self):
         """ Returns a LIST of queue parameters. Function first checks if ID exists, if so deletes and then checks the next queued until queue is empty (None) or queued id does not exist yet."""
         while True:
-            queueItem = self.download_queue.find_one({})
+            queueItem = self.download_queue.find_one({'status': 'queued'})
             if not queueItem:
                 return None
-            elif self.check_exists(self, queueItem['id']):
+            elif self.check_exists(queueItem['id']):
                 self.queue_deleteQueue(queueItem['id'])
+            self.download_queue.update_one({'id': queueItem['id']}, {'$set': {'status': 'working'}})
             return queueItem
         
-        
-        
 ##########################################
 #           HELPER FUNCTIONS             #
 ##########################################

ayta/tasks.py

@@ -42,6 +42,7 @@ def video_queue():
         get_nosql().queue_deleteQueue(videoId)
         return True
     else:
+        get_nosql().queue_setFailed(videoId)
         return False
 
 @shared_task()

ayta/templates/admin/queue.html

@@ -155,7 +155,7 @@
             </form>
             <form method="post">
               <input type="text" value="{{ id.get('id') }}" name="value" hidden>
-              <button class="btn-small waves-effect waves-light" type="submit" name="task" value="run-download" title="Run download task" disabled}>⏩</button>
+              <button class="btn-small waves-effect waves-light" type="submit" name="task" value="run-download" title="Run download task" disabled>⏩</button>
 			  <!-- This function fill not work until the download queue and video download process is rewritten -->
             </form>
           </td>