Merge pull request #327 from alphagov/add_job_row_number

Added job row number to the notification for csv jobs. All tests pass…

app/celery/tasks.py

@@ -148,17 +148,18 @@ def process_job(job_id):
         dao_get_template_by_id(job.template_id, job.template_version).__dict__
     )
 
-    for recipient, personalisation in RecipientCSV(
+    for row_number, recipient, personalisation in RecipientCSV(
             s3.get_job_from_s3(str(service.id), str(job_id)),
             template_type=template.template_type,
             placeholders=template.placeholders
-    ).recipients_and_personalisation:
+    ).enumerated_recipients_and_personalisation:
 
         encrypted = encryption.encrypt({
             'template': str(template.id),
             'template_version': job.template_version,
             'job': str(job.id),
             'to': recipient,
+            'row_number': row_number,
             'personalisation': {
                 key: personalisation.get(key)
                 for key in template.placeholders
@@ -242,6 +243,7 @@ def send_sms(service_id, notification_id, encrypted_notification, created_at):
             to=notification['to'],
             service_id=service_id,
             job_id=notification.get('job', None),
+            job_row_number=notification.get('row_number', None),
             status='failed' if restricted else 'sending',
             created_at=datetime.strptime(created_at, DATETIME_FORMAT),
             sent_at=sent_at,
@@ -307,6 +309,7 @@ def send_email(service_id, notification_id, from_address, encrypted_notification
             to=notification['to'],
             service_id=service_id,
             job_id=notification.get('job', None),
+            job_row_number=notification.get('row_number', None),
             status='failed' if restricted else 'sending',
             created_at=datetime.strptime(created_at, DATETIME_FORMAT),
             sent_at=sent_at,

app/dao/notifications_dao.py

@@ -1,4 +1,4 @@
-from sqlalchemy import (desc, func, Integer, and_)
+from sqlalchemy import (desc, func, Integer, and_, asc)
 from sqlalchemy.sql.expression import cast
 
 from datetime import (
@@ -227,7 +227,7 @@ def get_notifications_for_job(service_id, job_id, filter_dict=None, page=1, page
         page_size = current_app.config['PAGE_SIZE']
     query = Notification.query.filter_by(service_id=service_id, job_id=job_id)
     query = filter_query(query, filter_dict)
-    return query.order_by(desc(Notification.created_at)).paginate(
+    return query.order_by(asc(Notification.job_row_number)).paginate(
         page=page,
         per_page=page_size
     )

app/models.py

@@ -317,6 +317,7 @@ class Notification(db.Model):
     to = db.Column(db.String, nullable=False)
     job_id = db.Column(UUID(as_uuid=True), db.ForeignKey('jobs.id'), index=True, unique=False)
     job = db.relationship('Job', backref=db.backref('notifications', lazy='dynamic'))
+    job_row_number = db.Column(db.Integer, nullable=True)
     service_id = db.Column(UUID(as_uuid=True), db.ForeignKey('services.id'), index=True, unique=False)
     service = db.relationship('Service')
     template_id = db.Column(UUID(as_uuid=True), db.ForeignKey('templates.id'), index=True, unique=False)