Celery tests

app/__init__.py

@@ -11,12 +11,13 @@ from werkzeug.local import LocalProxy
 from config import configs
 from utils import logging
 from notify_client import NotifyAPIClient
+from app.celery.celery import NotifyCelery
 
 
 db = SQLAlchemy()
 ma = Marshmallow()
 notify_alpha_client = NotifyAPIClient()
-
+celery = NotifyCelery()
 api_user = LocalProxy(lambda: _request_ctx_stack.top.api_user)
 
 
@@ -32,6 +33,8 @@ def create_app(config_name, config_overrides=None):
     logging.init_app(application)
     notify_alpha_client.init_app(application)
 
+    celery.init_app(application)
+
     from app.service.rest import service as service_blueprint
     from app.user.rest import user as user_blueprint
     from app.template.rest import template as template_blueprint
@@ -75,6 +78,7 @@ def init_app(app, config_overrides):
         return response
 
 
+
 def convert_to_boolean(value):
     """Turn strings to bools if they look like them
 

app/celery/celery.py

@@ -0,0 +1,21 @@
+from celery import Celery
+
+
+class NotifyCelery(Celery):
+
+    def init_app(self, app):
+        super().__init__(app.import_name, broker=app.config['BROKER_URL'])
+        self.conf.update(app.config)
+        TaskBase = self.Task
+
+        class ContextTask(TaskBase):
+            abstract = True
+
+            def __call__(self, *args, **kwargs):
+                with app.app_context():
+                    return TaskBase.__call__(self, *args, **kwargs)
+        self.Task = ContextTask
+
+
+
+

app/celery/tasks.py

@@ -0,0 +1,9 @@
+from app import celery
+from app.dao.services_dao import get_model_services
+
+
+@celery.task(name="refresh-services")
+def refresh_services():
+    print(get_model_services())
+    for service in get_model_services():
+        celery.control.add_consumer(str(service.id))

app/notifications/rest.py

@@ -12,6 +12,7 @@ from app import notify_alpha_client
 from app import api_user
 from app.dao import (templates_dao, services_dao)
 import re
+from app import celery
 
 mobile_regex = re.compile("^\\+44[\\d]{10}$")
 
@@ -23,6 +24,15 @@ def get_notifications(notification_id):
     return jsonify(notify_alpha_client.fetch_notification_by_id(notification_id)), 200
 
 
+@celery.task(name="make-sms", bind="True")
+def send_sms(self, to, template):
+    print('Executing task id {0.id}, args: {0.args!r} kwargs: {0.kwargs!r}'.format(self.request))
+    from time import sleep
+    sleep(0.5)
+    print('finished')
+    #notify_alpha_client.send_sms(mobile_number=to, message=template)
+
+
 @notifications.route('/sms', methods=['POST'])
 def create_sms_notification():
     notification = request.get_json()['notification']
@@ -54,8 +64,9 @@ def create_sms_notification():
 
         # add notification to the queue
         service = services_dao.get_model_services(api_user['client'], _raise=False)
-        _add_notification_to_queue(template.id, service, 'sms', to)
+        #_add_notification_to_queue(template.id, service, 'sms', to)
-        return jsonify(notify_alpha_client.send_sms(mobile_number=to, message=template.content)), 200
+        send_sms.apply_async((to, template.content), queue=str(service.id))
+        return jsonify(success=True)  # notify_alpha_client.send_sms(mobile_number=to, message=template.content)), 200
 
 
 @notifications.route('/email', methods=['POST'])
@@ -151,3 +162,4 @@ def _add_notification_to_queue(template_id, service, msg_type, to):
                                       'message_id': {'StringValue': message_id, 'DataType': 'String'},
                                       'service_id': {'StringValue': str(service.id), 'DataType': 'String'},
                                       'template_id': {'StringValue': str(template_id), 'DataType': 'String'}})
+

config.py

@@ -1,4 +1,5 @@
 import os
+from datetime import timedelta
 
 
 class Config(object):
@@ -17,6 +18,25 @@ class Config(object):
     AWS_REGION = 'eu-west-1'
     NOTIFY_JOB_QUEUE = os.getenv('NOTIFY_JOB_QUEUE', 'notify-jobs-queue')
 
+    BROKER_URL = 'amqp://guest:guest@localhost:5672//'
+    BROKER_TRANSPORT_OPTIONS = {
+        'region': 'eu-west-1',
+        'polling_interval': 10,  # 1 second
+        'visibility_timeout': 3600,  # 1 hour
+        'queue_name_prefix': 'NOTIFY-CELERY-TEST-'
+    }
+    CELERY_ENABLE_UTC = True,
+    CELERY_TIMEZONE = 'Europe/London'
+    CELERY_ACCEPT_CONTENT = ['json']
+    CELERY_TASK_SERIALIZER = 'json'
+    CELERYBEAT_SCHEDULE = {
+        'refresh-queues': {
+            'task': 'refresh-services',
+            'schedule': timedelta(seconds=5)
+        }
+    }
+    CELERY_IMPORTS = ('app.celery.tasks',)
+
 
 class Development(Config):
     DEBUG = True
@@ -36,6 +56,10 @@ class Test(Config):
 
 
 class Live(Config):
+    SECRET_KEY = 'secret-key'
+    DANGEROUS_SALT = 'dangerous-salt'
+    ADMIN_CLIENT_USER_NAME = 'dev-notify-admin'
+    ADMIN_CLIENT_SECRET = 'dev-notify-secret-key'
     pass
 
 

requirements.txt

@@ -13,6 +13,8 @@ itsdangerous==0.24
 Flask-Bcrypt==0.6.2
 credstash==1.8.0
 boto3==1.2.3
+celery==3.1.20
+redis==2.10.5
 
 git+https://github.com/alphagov/notifications-python-client.git@0.2.1#egg=notifications-python-client==0.2.1
 

run_celery.py

@@ -0,0 +1,7 @@
+#!/usr/bin/env python
+import os
+from app import celery, create_app
+from app.celery.tasks import refresh_services
+
+application = create_app(os.getenv('NOTIFY_API_ENVIRONMENT') or 'development')
+application.app_context().push()

wsgi.py

@@ -1,9 +1,9 @@
 from app import create_app
 from credstash import getAllSecrets
 
-secrets = getAllSecrets(region="eu-west-1")
+#secrets = getAllSecrets(region="eu-west-1")
 
-application = create_app('live', secrets)
+application = create_app('live', None)
 
 if __name__ == "__main__":
         application.run()