Process Incomplete Jobs Refactor

- Moved the process_incomplete_jobs to tasks.py
- Moved the process_incomplete_jobs test to test_tasks.py
- Cleaned up imports and other code style issues.

As the new tasks is not a scheduled one, moved the the tasks.py file.
This makes it more consisted with other tasks. Updated a few code style
issues to make it more consistent with other coe and hence more
maintainable in future.

app/celery/tasks.py

@@ -2,7 +2,7 @@ import json
 from datetime import datetime
 from collections import namedtuple
 
-from celery.signals import worker_process_init, worker_process_shutdown
+from celery.signals import worker_process_shutdown
 from flask import current_app
 from notifications_utils.recipients import (
     RecipientCSV
@@ -44,9 +44,9 @@ from app.dao.service_inbound_api_dao import get_service_inbound_api_for_service
 from app.dao.services_dao import dao_fetch_service_by_id, fetch_todays_total_message_count
 from app.dao.templates_dao import dao_get_template_by_id
 from app.models import (
+    Job,
+    Notification,
     EMAIL_TYPE,
-    SMS_TYPE,
-    LETTER_TYPE,
     KEY_TYPE_NORMAL,
     JOB_STATUS_CANCELLED,
     JOB_STATUS_PENDING,
@@ -54,8 +54,10 @@ from app.models import (
     JOB_STATUS_FINISHED,
     JOB_STATUS_READY_TO_SEND,
     JOB_STATUS_SENT_TO_DVLA, JOB_STATUS_ERROR,
+    LETTER_TYPE,
     NOTIFICATION_SENDING,
-    NOTIFICATION_TECHNICAL_FAILURE
+    NOTIFICATION_TECHNICAL_FAILURE,
+    SMS_TYPE,
 )
 from app.notifications.process_notifications import persist_notification
 from app.service.utils import service_allowed_to_send_to
@@ -492,3 +494,44 @@ def send_inbound_sms_to_service(self, inbound_sms_id, service_id):
                                service_id, inbound_api.url, e))
             except self.MaxRetriesExceededError:
                 current_app.logger.exception('Retry: send_inbound_sms_to_service has retried the max number of times')
+
+
+@notify_celery.task(name='process-incomplete-jobs')
+@statsd(namespace="tasks")
+def process_incomplete_jobs(job_ids):
+    current_app.logger.info("Resuming Job(s) {}".format(job_ids))
+    for job_id in job_ids:
+        process_incomplete_job(job_id)
+
+
+def process_incomplete_job(job_id):
+
+    job = Job.query.filter(Job.id == job_id).one()
+
+    last_notification_added = Notification.query.filter(
+        Notification.job_id == job_id
+    ).order_by(
+        Notification.job_row_number.desc()
+    ).first()
+
+    if last_notification_added:
+        resume_from_row = last_notification_added.job_row_number
+    else:
+        resume_from_row = -1  # The first row in the csv with a number is row 0
+
+    current_app.logger.info("Resuming job {} from row {}".format(job_id, resume_from_row))
+
+    db_template = dao_get_template_by_id(job.template_id, job.template_version)
+
+    TemplateClass = get_template_class(db_template.template_type)
+    template = TemplateClass(db_template.__dict__)
+
+    for row_number, recipient, personalisation 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)
+
+    job_complete(job, job.service, template, True)