add endpoint for 'recent' notifications

app/dao/notifications_dao.py

@@ -267,6 +267,29 @@ def get_notifications_for_job(
     return pagination
 
 
+def get_recent_notifications_for_job(
+    service_id, job_id, filter_dict=None, page=1, page_size=None
+):
+    if page_size is None:
+        page_size = current_app.config["PAGE_SIZE"]
+
+    stmt = select(Notification).where(
+        Notification.service_id == service_id,
+        Notification.job_id == job_id,
+        Notification.status in [NotificationStatus.FAILED, Notification.DELIVERED],
+    )
+    stmt = _filter_query(stmt, filter_dict)
+    stmt = stmt.order_by(desc(Notification.job_row_number))
+
+    results = db.session.execute(stmt).scalars().all()
+
+    page_size = current_app.config["PAGE_SIZE"]
+    offset = (page - 1) * page_size
+    paginated_results = results[offset : offset + page_size]
+    pagination = Pagination(paginated_results, page, page_size, len(results))
+    return pagination
+
+
 def dao_get_notification_count_for_job_id(*, job_id):
     stmt = select(func.count(Notification.id)).where(Notification.job_id == job_id)
     return db.session.execute(stmt).scalar()

app/job/rest.py

@@ -22,6 +22,7 @@
 from app.dao.notifications_dao import (
     dao_get_notification_count_for_job_id,
     get_notifications_for_job,
+    get_recent_notifications_for_job,
 )
 from app.dao.services_dao import dao_fetch_service_by_id
 from app.dao.templates_dao import dao_get_template_by_id
@@ -124,6 +125,67 @@ def get_all_notifications_for_service_job(service_id, job_id):
     )
 
 
+@job_blueprint.route("/<job_id>/recent_notifications", methods=["GET"])
+def get_recent_notifications_for_service_job(service_id, job_id):
+    data = notifications_filter_schema.load(request.args)
+    page = data["page"] if "page" in data else 1
+    page_size = (
+        data["page_size"]
+        if "page_size" in data
+        else current_app.config.get("PAGE_SIZE")
+    )
+    paginated_notifications = get_recent_notifications_for_job(
+        service_id, job_id, filter_dict=data, page=page, page_size=page_size
+    )
+
+    kwargs = request.args.to_dict()
+    kwargs["service_id"] = service_id
+    kwargs["job_id"] = job_id
+
+    for notification in paginated_notifications.items:
+        if notification.job_id is not None:
+            recipient = get_phone_number_from_s3(
+                notification.service_id,
+                notification.job_id,
+                notification.job_row_number,
+            )
+            notification.to = recipient
+            notification.normalised_to = recipient
+
+    for notification in paginated_notifications.items:
+        if notification.job_id is not None:
+            notification.personalisation = get_personalisation_from_s3(
+                notification.service_id,
+                notification.job_id,
+                notification.job_row_number,
+            )
+
+    notifications = None
+    if data.get("format_for_csv"):
+        notifications = [
+            notification.serialize_for_csv()
+            for notification in paginated_notifications.items
+        ]
+    else:
+        notifications = notification_with_template_schema.dump(
+            paginated_notifications.items, many=True
+        )
+
+    return (
+        jsonify(
+            notifications=notifications,
+            page_size=page_size,
+            total=paginated_notifications.total,
+            links=pagination_links(
+                paginated_notifications,
+                ".get_all_notifications_for_service_job",
+                **kwargs,
+            ),
+        ),
+        200,
+    )
+
+
 @job_blueprint.route("/<job_id>/notification_count", methods=["GET"])
 def get_notification_count_for_job_id(service_id, job_id):
     dao_get_job_by_service_id_and_job_id(service_id, job_id)

app/service/rest.py

@@ -694,7 +694,12 @@ def get_single_month_notification_stats_by_user(service_id, user_id):
         service_id, start_date, end_date, user_id
     )
 
-    stats = get_specific_days_stats(results, start_date, end_date=end_date, total_notifications=total_notifications,)
+    stats = get_specific_days_stats(
+        results,
+        start_date,
+        end_date=end_date,
+        total_notifications=total_notifications,
+    )
     return jsonify(stats)