Count how many times a contact list has been used

Because we’ll be grouping jobs under their parent contact lists it will
be useful to have a way of showing how many times a contact list has
been used. This will give the right information scent to indicate that
clicking into a contact list is where you go to see its jobs. This means
that the API needs to return a count of jobs for each contact list.

Putting this code feels very non-idiomatic for our API. So suggestions
about how to better architect it welcome…

app/models.py

@@ -12,7 +12,7 @@ from sqlalchemy.dialects.postgresql import (
     JSONB,
 )
 from sqlalchemy.orm.collections import attribute_mapped_collection
-from sqlalchemy import UniqueConstraint, CheckConstraint, Index
+from sqlalchemy import UniqueConstraint, CheckConstraint, Index, String, and_, func
 from notifications_utils.columns import Columns
 from notifications_utils.recipients import (
     validate_email_address,
@@ -2138,12 +2138,27 @@ class ServiceContactList(db.Model):
     updated_at = db.Column(db.DateTime, nullable=True, onupdate=datetime.datetime.utcnow)
     archived = db.Column(db.Boolean, nullable=False, default=False)
 
+    def get_job_count(self):
+        today = datetime.datetime.utcnow().date()
+        return Job.query.filter(
+            Job.contact_list_id == self.id,
+            func.coalesce(
+                Job.processing_started, Job.created_at
+            ) >= today - func.coalesce(ServiceDataRetention.days_of_retention, 7)
+        ).outerjoin(
+            ServiceDataRetention, and_(
+                self.service_id == ServiceDataRetention.service_id,
+                func.cast(self.template_type, String) == func.cast(ServiceDataRetention.notification_type, String)
+            )
+        ).count()
+
     def serialize(self):
         created_at_in_bst = convert_utc_to_bst(self.created_at)
         contact_list = {
             "id": str(self.id),
             "original_file_name": self.original_file_name,
             "row_count": self.row_count,
+            "job_count": self.get_job_count(),
             "template_type": self.template_type,
             "service_id": str(self.service_id),
             "created_by": self.created_by.name,

tests/app/service/test_service_contact_list_rest.py

@@ -1,9 +1,13 @@
+import pytest
 import uuid
 
+from datetime import datetime, timedelta
+
 from app.models import ServiceContactList
 from tests.app.db import (
     create_job,
     create_service_contact_list,
+    create_service_data_retention,
     create_service,
     create_template,
 )
@@ -65,6 +69,48 @@ def test_get_contact_list(admin_request, notify_db_session):
 
     assert len(response) == 1
     assert response[0] == contact_list.serialize()
+    assert response[0]['job_count'] == 0
+
+
+@pytest.mark.parametrize('days_of_email_retention, expected_job_count', (
+    (None, 8),
+    (7, 8),
+    (3, 4),
+))
+def test_get_contact_list_counts_jobs(
+    sample_template,
+    admin_request,
+    days_of_email_retention,
+    expected_job_count,
+):
+    if days_of_email_retention:
+        create_service_data_retention(sample_template.service, 'email', days_of_email_retention)
+
+    # This should be ignored because it’s another template type
+    create_service_data_retention(sample_template.service, 'sms', 1)
+
+    contact_list_1 = create_service_contact_list(service=sample_template.service)
+    contact_list_2 = create_service_contact_list(service=sample_template.service)
+
+    for i in range(10):
+        create_job(
+            template=sample_template,
+            contact_list_id=contact_list_2.id,
+            created_at=datetime.utcnow() - timedelta(days=i)
+        )
+
+    response = admin_request.get(
+        'service.get_contact_list',
+        service_id=contact_list_1.service_id
+    )
+
+    assert len(response) == 2
+
+    assert response[0]['id'] == str(contact_list_2.id)
+    assert response[0]['job_count'] == expected_job_count
+
+    assert response[1]['id'] == str(contact_list_1.id)
+    assert response[1]['job_count'] == 0
 
 
 def test_get_contact_list_returns_for_service(admin_request, notify_db_session):