Bring in refactored CSV processing

Shouldn’t be any functional changes here, just things being named more
clearly.

app/celery/tasks.py

@@ -111,12 +111,12 @@ def process_job(job_id):
 
     current_app.logger.debug("Starting job {} processing {} notifications".format(job_id, job.notification_count))
 
-    for row_number, recipient, personalisation in RecipientCSV(
+    for row in RecipientCSV(
             s3.get_job_from_s3(str(service.id), str(job_id)),
             template_type=template.template_type,
             placeholders=template.placeholders
-    ).enumerated_recipients_and_personalisation:
-        process_row(row_number, recipient, personalisation, template, job, service)
+    ).rows:
+        process_row(row, template, job, service)
 
     job_complete(job, start=start)
 
@@ -138,15 +138,15 @@ def job_complete(job, resumed=False, start=None):
         )
 
 
-def process_row(row_number, recipient, personalisation, template, job, service):
+def process_row(row, template, job, service):
     template_type = template.template_type
     encrypted = encryption.encrypt({
         'template': str(template.id),
         'template_version': job.template_version,
         'job': str(job.id),
-        'to': recipient,
-        'row_number': row_number,
-        'personalisation': dict(personalisation)
+        'to': row.recipient,
+        'row_number': row.index,
+        'personalisation': dict(row.personalisation)
     })
 
     send_fns = {
@@ -573,12 +573,12 @@ def process_incomplete_job(job_id):
     TemplateClass = get_template_class(db_template.template_type)
     template = TemplateClass(db_template.__dict__)
 
-    for row_number, recipient, personalisation in RecipientCSV(
+    for row in RecipientCSV(
             s3.get_job_from_s3(str(job.service_id), str(job.id)),
             template_type=template.template_type,
             placeholders=template.placeholders
-    ).enumerated_recipients_and_personalisation:
-        if row_number > resume_from_row:
-            process_row(row_number, recipient, personalisation, template, job, job.service)
+    ).rows:
+        if row.index > resume_from_row:
+            process_row(row, template, job, job.service)
 
     job_complete(job, resumed=True)

requirements.txt

@@ -23,6 +23,6 @@ notifications-python-client==4.7.2
 # PaaS
 awscli-cwlogs>=1.4,<1.5
 
-git+https://github.com/alphagov/notifications-utils.git@24.0.0#egg=notifications-utils==24.0.0
+git+https://github.com/alphagov/notifications-utils.git@25.0.0#egg=notifications-utils==25.0.0
 
 git+https://github.com/alphagov/boto.git@2.43.0-patch3#egg=boto==2.43.0-patch3

tests/app/celery/test_tasks.py

@@ -10,6 +10,7 @@ from requests import RequestException
 from sqlalchemy.exc import SQLAlchemyError
 from celery.exceptions import Retry
 from notifications_utils.template import SMSMessageTemplate, WithSubjectTemplate
+from notifications_utils.columns import Row
 
 from app import (encryption, DATETIME_FORMAT)
 from app.celery import provider_tasks
@@ -303,18 +304,17 @@ def test_should_process_letter_job(sample_letter_job, mocker):
     )
 
     row_call = process_row_mock.mock_calls[0][1]
-
-    assert row_call[0] == 0
-    assert row_call[1] == ['A1', 'A2', 'A3', 'A4', None, None, 'A_POST']
-    assert dict(row_call[2]) == {
+    assert row_call[0].index == 0
+    assert row_call[0].recipient == ['A1', 'A2', 'A3', 'A4', None, None, 'A_POST']
+    assert row_call[0].personalisation == {
         'addressline1': 'A1',
         'addressline2': 'A2',
         'addressline3': 'A3',
         'addressline4': 'A4',
         'postcode': 'A_POST'
     }
-    assert row_call[4] == sample_letter_job
-    assert row_call[5] == sample_letter_job.service
+    assert row_call[2] == sample_letter_job
+    assert row_call[3] == sample_letter_job.service
 
     assert process_row_mock.call_count == 1
 
@@ -363,7 +363,19 @@ def test_process_row_sends_letter_task(template_type, research_mode, expected_fu
     job = Mock(id='job_id', template_version='temp_vers')
     service = Mock(id='service_id', research_mode=research_mode)
 
-    process_row('row_num', 'recip', {'foo': 'bar'}, template, job, service)
+    process_row(
+        Row(
+            {'foo': 'bar', 'to': 'recip'},
+            index='row_num',
+            error_fn=lambda k, v: None,
+            recipient_column_headers=['to'],
+            placeholders={'foo'},
+            template=template,
+        ),
+        template,
+        job,
+        service,
+    )
 
     encrypt_mock.assert_called_once_with({
         'template': 'template_id',