mirror of
https://github.com/GSA/notifications-api.git
synced 2026-09-04 06:48:26 -04:00
832005efef
- previously this was unbounded, so it got all jobs older then 7 days. In excess of 75,000 🔥
- this meant that the job took (a) a long time and (b) a lot memory and (c) doing the same thing every day
These changes mean that the job has a 2 day eligible window for jobs, minimising the number of eligible jobs in a run, whilst still retaining some leeway in event if it failing one night.
In principle the job runs early morning on a given day. The previous 7 days are left along, and then the previous 2 days worth of files are deleted:
so:
runs on
31st
30,29,28,27,26,25,24 are ignored
23,22 jobs here have files deleted
21 and earlier are ignored.
124 lines
3.7 KiB
Python
124 lines
3.7 KiB
Python
from datetime import datetime
|
|
|
|
from flask import current_app
|
|
from sqlalchemy import func, desc, asc, cast, Date as sql_date
|
|
|
|
from app import db
|
|
from app.dao import days_ago
|
|
from app.models import (Job,
|
|
Notification,
|
|
NotificationHistory,
|
|
JOB_STATUS_SCHEDULED,
|
|
JOB_STATUS_PENDING)
|
|
from app.statsd_decorators import statsd
|
|
|
|
|
|
@statsd(namespace="dao")
|
|
def dao_get_notification_outcomes_for_job(service_id, job_id):
|
|
query = db.session.query(
|
|
func.count(NotificationHistory.status).label('count'),
|
|
NotificationHistory.status.label('status')
|
|
)
|
|
|
|
return query \
|
|
.filter(NotificationHistory.service_id == service_id) \
|
|
.filter(NotificationHistory.job_id == job_id)\
|
|
.group_by(NotificationHistory.status) \
|
|
.order_by(asc(NotificationHistory.status)) \
|
|
.all()
|
|
|
|
|
|
@statsd(namespace="dao")
|
|
def all_notifications_are_created_for_job(job_id):
|
|
query = db.session.query(func.count(Notification.id), Job.id)\
|
|
.join(Job)\
|
|
.filter(Job.id == job_id)\
|
|
.group_by(Job.id)\
|
|
.having(func.count(Notification.id) == Job.notification_count).all()
|
|
|
|
return query
|
|
|
|
|
|
@statsd(namespace="dao")
|
|
def dao_get_all_notifications_for_job(job_id):
|
|
return db.session.query(Notification).filter(Notification.job_id == job_id).all()
|
|
|
|
|
|
def dao_get_job_by_service_id_and_job_id(service_id, job_id):
|
|
return Job.query.filter_by(service_id=service_id, id=job_id).one()
|
|
|
|
|
|
def dao_get_jobs_by_service_id(service_id, limit_days=None, page=1, page_size=50, statuses=None):
|
|
query_filter = [
|
|
Job.service_id == service_id,
|
|
Job.original_file_name != current_app.config['TEST_MESSAGE_FILENAME']
|
|
]
|
|
if limit_days is not None:
|
|
query_filter.append(cast(Job.created_at, sql_date) >= days_ago(limit_days))
|
|
if statuses is not None and statuses != ['']:
|
|
query_filter.append(
|
|
Job.job_status.in_(statuses)
|
|
)
|
|
return Job.query \
|
|
.filter(*query_filter) \
|
|
.order_by(Job.processing_started.desc(), Job.created_at.desc()) \
|
|
.paginate(page=page, per_page=page_size)
|
|
|
|
|
|
def dao_get_job_by_id(job_id):
|
|
return Job.query.filter_by(id=job_id).one()
|
|
|
|
|
|
def dao_set_scheduled_jobs_to_pending():
|
|
"""
|
|
Sets all past scheduled jobs to pending, and then returns them for further processing.
|
|
|
|
this is used in the run_scheduled_jobs task, so we put a FOR UPDATE lock on the job table for the duration of
|
|
the transaction so that if the task is run more than once concurrently, one task will block the other select
|
|
from completing until it commits.
|
|
"""
|
|
jobs = Job.query \
|
|
.filter(
|
|
Job.job_status == JOB_STATUS_SCHEDULED,
|
|
Job.scheduled_for < datetime.utcnow()
|
|
) \
|
|
.order_by(asc(Job.scheduled_for)) \
|
|
.with_for_update() \
|
|
.all()
|
|
|
|
for job in jobs:
|
|
job.job_status = JOB_STATUS_PENDING
|
|
|
|
db.session.add_all(jobs)
|
|
db.session.commit()
|
|
|
|
return jobs
|
|
|
|
|
|
def dao_get_future_scheduled_job_by_id_and_service_id(job_id, service_id):
|
|
return Job.query \
|
|
.filter(
|
|
Job.service_id == service_id,
|
|
Job.id == job_id,
|
|
Job.job_status == JOB_STATUS_SCHEDULED,
|
|
Job.scheduled_for > datetime.utcnow()
|
|
) \
|
|
.one()
|
|
|
|
|
|
def dao_create_job(job):
|
|
db.session.add(job)
|
|
db.session.commit()
|
|
|
|
|
|
def dao_update_job(job):
|
|
db.session.add(job)
|
|
db.session.commit()
|
|
|
|
|
|
def dao_get_jobs_older_than_limited_by(older_than=7, limit_days=2):
|
|
return Job.query.filter(
|
|
cast(Job.created_at, sql_date) < days_ago(older_than),
|
|
cast(Job.created_at, sql_date) >= days_ago(older_than + limit_days)
|
|
).order_by(desc(Job.created_at)).all()
|