Renamed the header of the CSV to 'to' from 'number' to allow for email jobs

- added new columns to Job and Notification to capture the start/end dates accurately

app/celery/tasks.py

@@ -3,13 +3,14 @@ from app import notify_celery, encryption, firetext_client, aws_ses_client
 from app.clients.email.aws_ses import AwsSesClientException
 from app.clients.sms.firetext import FiretextClientException
 from app.dao.templates_dao import dao_get_template_by_id
-from app.dao.notifications_dao import save_notification
+from app.dao.notifications_dao import dao_create_notification, dao_update_notification
 from app.dao.jobs_dao import dao_update_job, dao_get_job_by_id
 from app.models import Notification
 from flask import current_app
 from sqlalchemy.exc import SQLAlchemyError
 from app.aws import s3
-from app.csv import get_mobile_numbers_from_csv
+from app.csv import get_recipient_from_csv
+from datetime import datetime
 
 
 @notify_celery.task(name="process-job")
@@ -19,28 +20,38 @@ def process_job(job_id):
     dao_update_job(job)
 
     file = s3.get_job_from_s3(job.bucket_name, job_id)
-    mobile_numbers = get_mobile_numbers_from_csv(file)
+    recipients = get_recipient_from_csv(file)
 
-    for mobile_number in mobile_numbers:
-        notification = encryption.encrypt({
+    for recipient in recipients:
+        encrypted = encryption.encrypt({
             'template': job.template_id,
             'job': str(job.id),
-            'to': mobile_number
+            'to': recipient
         })
 
-        send_sms.apply_async((
-            str(job.service_id),
-            str(create_uuid()),
-            notification),
-            queue='sms'
-        )
+        if job.template.template_type == 'sms':
+            send_sms.apply_async((
+                str(job.service_id),
+                str(create_uuid()),
+                encrypted),
+                queue='bulk-sms'
+            )
+
+        if job.template.template_type == 'email':
+            send_email.apply_async((
+                str(job.service_id),
+                str(create_uuid()),
+                job.template.subject,
+                "{}@{}".format(job.service.email_from, current_app.config['NOTIFY_EMAIL_DOMAIN']),
+                encrypted),
+                queue='bulk-email')
 
     job.status = 'finished'
     dao_update_job(job)
 
 
 @notify_celery.task(name="send-sms")
-def send_sms(service_id, notification_id, encrypted_notification):
+def send_sms(service_id, notification_id, encrypted_notification, created_at):
     notification = encryption.decrypt(encrypted_notification)
     template = dao_get_template_by_id(notification['template'])
 
@@ -51,22 +62,25 @@ def send_sms(service_id, notification_id, encrypted_notification):
             to=notification['to'],
             service_id=service_id,
             job_id=notification.get('job', None),
-            status='sent'
+            status='sent',
+            created_at=created_at,
+            sent_at=datetime.utcnow()
         )
-        save_notification(notification_db_object)
+        dao_create_notification(notification_db_object)
 
         try:
             firetext_client.send_sms(notification['to'], template.content)
         except FiretextClientException as e:
             current_app.logger.debug(e)
-            save_notification(notification_db_object, {"status": "failed"})
+            notification_db_object.status = 'failed'
+            dao_update_notification(notification_db_object)
 
     except SQLAlchemyError as e:
         current_app.logger.debug(e)
 
 
 @notify_celery.task(name="send-email")
-def send_email(service_id, notification_id, subject, from_address, encrypted_notification):
+def send_email(service_id, notification_id, subject, from_address, encrypted_notification, created_at):
     notification = encryption.decrypt(encrypted_notification)
     template = dao_get_template_by_id(notification['template'])
 
@@ -77,9 +91,11 @@ def send_email(service_id, notification_id, subject, from_address, encrypted_not
             to=notification['to'],
             service_id=service_id,
             job_id=notification.get('job', None),
-            status='sent'
+            status='sent',
+            created_at=created_at,
+            sent_at=datetime.utcnow()
         )
-        save_notification(notification_db_object)
+        dao_create_notification(notification_db_object)
 
         try:
             aws_ses_client.send_email(
@@ -90,7 +106,8 @@ def send_email(service_id, notification_id, subject, from_address, encrypted_not
             )
         except AwsSesClientException as e:
             current_app.logger.debug(e)
-            save_notification(notification_db_object, {"status": "failed"})
+            notification_db_object.status = 'failed'
+            dao_update_notification(notification_db_object)
 
     except SQLAlchemyError as e:
         current_app.logger.debug(e)

app/csv.py

@@ -1,12 +1,12 @@
 import csv
 
 
-def get_mobile_numbers_from_csv(file_data):
+def get_recipient_from_csv(file_data):
     numbers = []
     reader = csv.DictReader(
         file_data.splitlines(),
         lineterminator='\n',
         quoting=csv.QUOTE_NONE)
     for i, row in enumerate(reader):
-        numbers.append(row['phone'].replace(' ', ''))
+        numbers.append(row['to'].replace(' ', ''))
     return numbers

app/dao/notifications_dao.py

@@ -2,15 +2,13 @@ from app import db
 from app.models import Notification
 
 
-def save_notification(notification, update_dict={}):
-    if update_dict:
-        update_dict.pop('id', None)
-        update_dict.pop('job', None)
-        update_dict.pop('service', None)
-        update_dict.pop('template', None)
-        Notification.query.filter_by(id=notification.id).update(update_dict)
-    else:
-        db.session.add(notification)
+def dao_create_notification(notification):
+    db.session.add(notification)
+    db.session.commit()
+
+
+def dao_update_notification(notification):
+    db.session.add(notification)
     db.session.commit()
 
 

app/models.py

@@ -160,6 +160,16 @@ class Job(db.Model):
         onupdate=datetime.datetime.now)
     status = db.Column(db.Enum(*JOB_STATUS_TYPES, name='job_status_types'), nullable=False, default='pending')
     notification_count = db.Column(db.Integer, nullable=False)
+    processing_started = db.Column(
+        db.DateTime,
+        index=False,
+        unique=False,
+        nullable=True)
+    processing_finished = db.Column(
+        db.DateTime,
+        index=False,
+        unique=False,
+        nullable=True)
 
 
 VERIFY_CODE_TYPES = ['email', 'sms']
@@ -214,8 +224,13 @@ class Notification(db.Model):
         db.DateTime,
         index=False,
         unique=False,
-        nullable=False,
-        default=datetime.datetime.now)
+        nullable=False)
+    sent_at = db.Column(
+        db.DateTime,
+        index=False,
+        unique=False,
+        nullable=True)
+    sent_by = db.Column(db.String, nullable=True)
     updated_at = db.Column(
         db.DateTime,
         index=False,