diff --git a/app/dao/jobs_dao.py b/app/dao/jobs_dao.py
index 13742418a..9d68337cd 100644
--- a/app/dao/jobs_dao.py
+++ b/app/dao/jobs_dao.py
@@ -1,6 +1,6 @@
-from datetime import date, timedelta
+from datetime import date, timedelta, datetime
 
-from sqlalchemy import desc, cast, Date as sql_date
+from sqlalchemy import desc, asc, cast, Date as sql_date
 from app import db
 from app.dao import days_ago
 from app.models import Job
@@ -21,6 +21,13 @@ def dao_get_job_by_id(job_id):
     return Job.query.filter_by(id=job_id).one()
 
 
+def dao_get_scheduled_jobs():
+    return Job.query \
+        .filter(Job.job_status == 'scheduled', Job.scheduled_for < datetime.utcnow()) \
+        .order_by(asc(Job.scheduled_for)) \
+        .all()
+
+
 def dao_create_job(job):
     db.session.add(job)
     db.session.commit()
diff --git a/tests/app/dao/test_jobs_dao.py b/tests/app/dao/test_jobs_dao.py
index 9f265ab0d..0d13d5fe9 100644
--- a/tests/app/dao/test_jobs_dao.py
+++ b/tests/app/dao/test_jobs_dao.py
@@ -5,10 +5,12 @@ from app.dao.jobs_dao import (
     dao_get_job_by_service_id_and_job_id,
     dao_create_job,
     dao_update_job,
-    dao_get_jobs_by_service_id
+    dao_get_jobs_by_service_id,
+    dao_get_scheduled_jobs
 )
 
 from app.models import Job
+from tests.app.conftest import sample_job
 
 
 def test_create_job(sample_template):
@@ -137,3 +139,30 @@ def test_update_job(sample_job):
     job_from_db = Job.query.get(sample_job.id)
 
     assert job_from_db.status == 'in progress'
+
+
+def test_get_scheduled_jobs_gets_all_jobs_in_scheduled_state_scheduled_before_now(notify_db, notify_db_session):
+    one_minute_ago = datetime.utcnow() - timedelta(minutes=1)
+    one_hour_ago = datetime.utcnow() - timedelta(minutes=60)
+    job_new = sample_job(notify_db, notify_db_session, scheduled_for=one_minute_ago, job_status='scheduled')
+    job_old = sample_job(notify_db, notify_db_session, scheduled_for=one_hour_ago, job_status='scheduled')
+    jobs = dao_get_scheduled_jobs()
+    assert len(jobs) == 2
+    assert jobs[0].id == job_old.id
+    assert jobs[1].id == job_new.id
+
+
+def test_get_scheduled_jobs_gets_ignores_jobs_not_scheduled(notify_db, notify_db_session):
+    one_minute_ago = datetime.utcnow() - timedelta(minutes=1)
+    sample_job(notify_db, notify_db_session)
+    job_scheduled = sample_job(notify_db, notify_db_session, scheduled_for=one_minute_ago, job_status='scheduled')
+    jobs = dao_get_scheduled_jobs()
+    assert len(jobs) == 1
+    assert jobs[0].id == job_scheduled.id
+
+
+def test_get_scheduled_jobs_gets_ignores_jobs_scheduled_in_the_future(notify_db, notify_db_session):
+    one_minute_in_the_future = datetime.utcnow() + timedelta(minutes=1)
+    sample_job(notify_db, notify_db_session, scheduled_for=one_minute_in_the_future, job_status='scheduled')
+    jobs = dao_get_scheduled_jobs()
+    assert len(jobs) == 0