WIP - task reports

app/services/spawn.py

@@ -4,7 +4,7 @@
 from app.libs.logger import logger
 from app.repository.singleton import Singleton
 from app.services.maps.spawn_map import SpawnMap
-from app.tasks import task_connections, task_webhook, task_counter
+from app.tasks import task_connections, task_webhook, task_counter, task_reports
 
 
 class SpawnJobs(object, metaclass=Singleton):
@@ -59,9 +59,10 @@ def is_disabled(sjob):
 
     @staticmethod
     def caller(task, args):
-        tasks = {'connections': task_connections, 'webhook': task_webhook}
+        tasks = {'connections': task_connections, 'webhook': task_webhook, 'reports': task_reports}
 
         if task in tasks:
+
             task_id = tasks[task].delay(**args)
             counter_id = task_counter.delay(_id=args.get('_id'))
-            logger.info('Scheduler: Task executed %s (%s)', task_id, counter_id)
+            logger.info('Scheduler: Task executed %s (%s)', task_id, counter_id)

app/tasks/__init__.py

@@ -1,6 +1,7 @@
 from .notify_event import task_notify_event
 from .webhook import task_webhook
 from .connections import task_connections
+from .reports import task_reports
 from .counter import task_counter
 from .crawling_job import task_crawling
 from .depleted_job import task_deplete

app/tasks/chain_exec.py

@@ -5,9 +5,11 @@
 def task_chain_exec(task, args):
     import app.tasks.webhook as module_webhook
     import app.tasks.connections as module_connections
+    import app.tasks.reports as module_reports
 
     tasks = {
         'connections': module_connections.task_connections,
+        'reports': module_reports.task_reports,
         'webhook': module_webhook.task_webhook
     }
 

app/tasks/reports.py

@@ -0,0 +1,24 @@
+import re
+import requests
+import json
+
+from app import celery
+
+from app.libs.url import FactoryURL
+from app.tasks.webhook import task_webhook
+from app.tasks.depleted_job import task_deplete
+
+
+@celery.task(name="reports")
+def task_reports(name, _id, endpoint, method="GET", args=[], params={}, chain=[]):
+    print(name, _id, endpoint, method, args, params, chain)
+    query = json.dumps({
+        '_id': _id,
+        'active': True
+    })
+
+    path = FactoryURL.make(path="reports")
+    resource = requests.post(path, json={'query': query})
+
+
+    return {_id}

package.json

@@ -5,7 +5,7 @@
   "main": "run.py",
   "scripts": {
     "beat": "python run.py",
-    "worker": "celery -A app.celery worker --loglevel=info",
+    "worker": "celery -A app.celery worker -E -Q scheduler --hostname=scheduler@%h --loglevel=info",
     "save": "pip freeze -l > requirements.txt ",
     "test": "python -m unittest discover",
     "package-flask": "docker build -t scheduler-maestro .",