Importing data asynchronously

Signed-off-by: Sergio Oliveira <sergio@tracy.com.br>

Importing data asynchronously
Signed-off-by: Sergio Oliveira <sergio@tracy.com.br>
Sergio Oliveira
1 parent 4a4ddcb1
Showing 6 changed files with 60 additions and 1 deletions Show diff stats
colab/plugins/apps.py
colab/plugins/tasks.py
colab/plugins/utils/data.py
colab/plugins/utils/proxy_data_api.py
vagrant/misc/etc/default/celerybeat
vagrant/provision.sh
  
 from django.apps import AppConfig
  
+from .utils.data import register_tasks
 from .utils.signals import connect_signal, register_signal
  
  
@@ -10,3 +11,5 @@ class PluginAppConfig(AppConfig):
     def ready(self):
         register_signal()
         connect_signal()
+
+        register_tasks()
@@ -0,0 +1,11 @@
+
+from datetime import timedelta
+from celery.decorators import periodic_task
+
+from .utils import data
+
+
+@periodic_task(run_every=timedelta(minutes=1))
+def import_plugin_data():
+    for task in data.TASKS:
+        task.delay()
@@ -0,0 +1,40 @@
+#!/usr/bin/env python
+
+import importlib
+
+from django.conf import settings
+
+from colab.celery import app
+from proxy_data_api import ProxyDataAPI
+
+
+TASKS = set()
+
+
+def register_tasks():
+
+    global TASKS
+
+    for app_name in settings.INSTALLED_APPS:
+
+        try:
+            module = importlib.import_module('{}.data_api'.format(app_name))
+        except ImportError as e:
+            continue
+
+        for item_name in dir(module):
+            item = getattr(module, item_name)
+            if item is ProxyDataAPI:
+                continue
+
+            if callable(getattr(item, 'fetch_data', None)):
+                instance = item()
+                task = app.task(bind=True)(instance.fetch_data)
+                TASKS.add(task)
+
+    return TASKS
+
+
+def data_import(self):
+    for task in TASKS:
+        task.delay()
  
+import abc
  
 class ProxyDataAPI(object):
  
+    @abc.abstractmethod
     def fetch_data(self):
         raise NotImplementedError('fetchData not yet implemented')
@@ -15,7 +15,7 @@ CELERTBEAT_LOG_FILE=&quot;/var/log/celery/beat.log&quot;
 CELERYBEAT_PID_FILE="/var/run/celery/beat.pid"
  
 CELERYBEAT_USER="vagrant"
-CELERYBEAT_GROUP="CELERYBEAT_GROUP"
+CELERYBEAT_GROUP="vagrant"
  
 # If enabled pid and log directories will be created if missing,
 # and owned by the userid/group configured.
@@ -45,6 +45,9 @@ colab-admin loaddata /vagrant/tests/test_data.json
 # Init.d Celery files
 sudo cp $basedir/vagrant/misc/etc/init.d/celery* /etc/init.d/
 sudo cp $basedir/vagrant/misc/etc/default/celery* /etc/default/
+sudo service celeryd stop || echo
+sudo service celerybeat stop || echo
+sleep 2
 sudo service celeryd start
 sudo service celerybeat start
1	1
2	2	from django.apps import AppConfig
3	3
	4	+from .utils.data import register_tasks
4	5	from .utils.signals import connect_signal, register_signal
5	6
6	7
...	...	@@ -10,3 +11,5 @@ class PluginAppConfig(AppConfig):
10	11	def ready(self):
11	12	register_signal()
12	13	connect_signal()
	14	+
	15	+ register_tasks()
...	...
...	...	@@ -0,0 +1,11 @@
	1	+
	2	+from datetime import timedelta
	3	+from celery.decorators import periodic_task
	4	+
	5	+from .utils import data
	6	+
	7	+
	8	+@periodic_task(run_every=timedelta(minutes=1))
	9	+def import_plugin_data():
	10	+ for task in data.TASKS:
	11	+ task.delay()
...	...
...	...	@@ -0,0 +1,40 @@
	1	+#!/usr/bin/env python
	2	+
	3	+import importlib
	4	+
	5	+from django.conf import settings
	6	+
	7	+from colab.celery import app
	8	+from proxy_data_api import ProxyDataAPI
	9	+
	10	+
	11	+TASKS = set()
	12	+
	13	+
	14	+def register_tasks():
	15	+
	16	+ global TASKS
	17	+
	18	+ for app_name in settings.INSTALLED_APPS:
	19	+
	20	+ try:
	21	+ module = importlib.import_module('{}.data_api'.format(app_name))
	22	+ except ImportError as e:
	23	+ continue
	24	+
	25	+ for item_name in dir(module):
	26	+ item = getattr(module, item_name)
	27	+ if item is ProxyDataAPI:
	28	+ continue
	29	+
	30	+ if callable(getattr(item, 'fetch_data', None)):
	31	+ instance = item()
	32	+ task = app.task(bind=True)(instance.fetch_data)
	33	+ TASKS.add(task)
	34	+
	35	+ return TASKS
	36	+
	37	+
	38	+def data_import(self):
	39	+ for task in TASKS:
	40	+ task.delay()
...	...
1	1
	2	+import abc
2	3
3	4	class ProxyDataAPI(object):
4	5
	6	+ @abc.abstractmethod
5	7	def fetch_data(self):
6	8	raise NotImplementedError('fetchData not yet implemented')
...	...
...	...	@@ -15,7 +15,7 @@ CELERTBEAT_LOG_FILE="/var/log/celery/beat.log"
15	15	CELERYBEAT_PID_FILE="/var/run/celery/beat.pid"
16	16
17	17	CELERYBEAT_USER="vagrant"
18		-CELERYBEAT_GROUP="CELERYBEAT_GROUP"
	18	+CELERYBEAT_GROUP="vagrant"
19	19
20	20	# If enabled pid and log directories will be created if missing,
21	21	# and owned by the userid/group configured.
...	...
...	...	@@ -45,6 +45,9 @@ colab-admin loaddata /vagrant/tests/test_data.json
45	45	# Init.d Celery files
46	46	sudo cp $basedir/vagrant/misc/etc/init.d/celery* /etc/init.d/
47	47	sudo cp $basedir/vagrant/misc/etc/default/celery* /etc/default/
	48	+sudo service celeryd stop \|\| echo
	49	+sudo service celerybeat stop \|\| echo
	50	+sleep 2
48	51	sudo service celeryd start
49	52	sudo service celerybeat start
50	53
...	...