darkhelm/notifications-api
1
0
Fork 0
mirror of https://github.com/GSA/notifications-api.git synced 2025-12-12 08:12:27 -05:00
Code Issues Packages Projects Releases Wiki Activity

Job Status changed.

Browse Source 
Signed-off-by: Cliff Hill <Clifford.hill@gsa.gov>
This commit is contained in:
Cliff Hill
2024-01-15 14:22:56 -05:00
parent 69a9accfca
commit db3761609b
7 changed files with 27 additions and 64 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
app/celery/scheduled_tasks.py
View File

@@ -34,8 +34,8 @@ from app.dao.services_dao import (
)
from app.dao.users_dao import delete_codes_older_created_more_than_a_day_ago
from app.delivery.send_to_providers import provider_to_use
from app.enums import NotificationType
from app.models import JOB_STATUS_ERROR, JOB_STATUS_IN_PROGRESS, JOB_STATUS_PENDING, Job
from app.enums import NotificationType, JobStatus
from app.models import Job
from app.notifications.process_notifications import send_notification_to_queue
MAX_NOTIFICATION_FAILS = 10000
@@ -186,11 +186,11 @@ def check_job_status():
thirty_five_minutes_ago = datetime.utcnow() - timedelta(minutes=35)
incomplete_in_progress_jobs = Job.query.filter(
Job.job_status == JOB_STATUS_IN_PROGRESS,
Job.job_status == JobStatus.IN_PROGRESS,
between(Job.processing_started, thirty_five_minutes_ago, thirty_minutes_ago),
)
incomplete_pending_jobs = Job.query.filter(
Job.job_status == JOB_STATUS_PENDING,
Job.job_status == JobStatus.PENDING,
Job.scheduled_for.isnot(None),
between(Job.scheduled_for, thirty_five_minutes_ago, thirty_minutes_ago),
)
@@ -205,7 +205,7 @@ def check_job_status():
# if they haven't been re-processed in time.
job_ids = []
for job in jobs_not_complete_after_30_minutes:
job.job_status = JOB_STATUS_ERROR
job.job_status = JobStatus.ERROR
dao_update_job(job)
job_ids.append(str(job.id))

20
app/celery/tasks.py
View File

@@ -20,14 +20,8 @@ from app.dao.service_email_reply_to_dao import dao_get_reply_to_by_id
from app.dao.service_inbound_api_dao import get_service_inbound_api_for_service
from app.dao.service_sms_sender_dao import dao_get_service_sms_senders_by_id
from app.dao.templates_dao import dao_get_template_by_id
from app.enums import NotificationType
from app.models import (
JOB_STATUS_CANCELLED,
JOB_STATUS_FINISHED,
JOB_STATUS_IN_PROGRESS,
JOB_STATUS_PENDING,
KEY_TYPE_NORMAL,
)
from app.enums import NotificationType, JobStatus
from app.models import KEY_TYPE_NORMAL
from app.notifications.process_notifications import persist_notification
from app.notifications.validators import check_service_over_total_message_limit
from app.serialised_models import SerialisedService, SerialisedTemplate
@@ -46,17 +40,17 @@ def process_job(job_id, sender_id=None):
)
)
if job.job_status != JOB_STATUS_PENDING:
if job.job_status != JobStatus.PENDING:
return
service = job.service
job.job_status = JOB_STATUS_IN_PROGRESS
job.job_status = JobStatus.IN_PROGRESS
job.processing_started = start
dao_update_job(job)
if not service.active:
job.job_status = JOB_STATUS_CANCELLED
job.job_status = JobStatus.CANCELLED
dao_update_job(job)
current_app.logger.warning(
"Job {} has been cancelled, service {} is inactive".format(
@@ -85,7 +79,7 @@ def process_job(job_id, sender_id=None):
def job_complete(job, resumed=False, start=None):
job.job_status = JOB_STATUS_FINISHED
job.job_status = JobStatus.FINISHED
finished = datetime.utcnow()
job.processing_finished = finished
@@ -432,7 +426,7 @@ def process_incomplete_jobs(job_ids):
# reset the processing start time so that the check_job_status scheduled task doesn't pick this job up again
for job in jobs:
job.job_status = JOB_STATUS_IN_PROGRESS
job.job_status = JobStatus.IN_PROGRESS
job.processing_started = datetime.utcnow()
dao_update_job(job)

14
app/dao/jobs_dao.py
View File

@@ -5,10 +5,8 @@ from flask import current_app
from sqlalchemy import and_, asc, desc, func
from app import db
from app.enums import JobStatus
from app.models import (
JOB_STATUS_FINISHED,
JOB_STATUS_PENDING,
JOB_STATUS_SCHEDULED,
FactNotificationStatus,
Job,
Notification,
@@ -85,7 +83,7 @@ def dao_get_scheduled_job_stats(
)
.filter(
Job.service_id == service_id,
Job.job_status == JOB_STATUS_SCHEDULED,
Job.job_status == JobStatus.SCHEDULED,
)
.one()
)
@@ -111,7 +109,7 @@ def dao_set_scheduled_jobs_to_pending():
"""
jobs = (
Job.query.filter(
Job.job_status == JOB_STATUS_SCHEDULED,
Job.job_status == JobStatus.SCHEDULED,
Job.scheduled_for < datetime.utcnow(),
)
.order_by(asc(Job.scheduled_for))
@@ -120,7 +118,7 @@ def dao_set_scheduled_jobs_to_pending():
)
for job in jobs:
job.job_status = JOB_STATUS_PENDING
job.job_status = JobStatus.PENDING
db.session.add_all(jobs)
db.session.commit()
@@ -132,7 +130,7 @@ 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.job_status == JobStatus.SCHEDULED,
Job.scheduled_for > datetime.utcnow(),
).one()
@@ -200,7 +198,7 @@ def find_jobs_with_missing_rows():
jobs_with_rows_missing = (
db.session.query(Job)
.filter(
Job.job_status == JOB_STATUS_FINISHED,
Job.job_status == JobStatus.FINISHED,
Job.processing_finished < ten_minutes_ago,
Job.processing_finished > yesterday,
Job.id == Notification.job_id,

6
app/dao/uploads_dao.py
View File

@@ -5,10 +5,8 @@ from flask import current_app
from sqlalchemy import String, and_, desc, func, literal, text
from app import db
from app.enums import NotificationStatus, NotificationType
from app.enums import NotificationStatus, NotificationType, JobStatus
from app.models import (
JOB_STATUS_CANCELLED,
JOB_STATUS_SCHEDULED,
Job,
Notification,
ServiceDataRetention,
@@ -52,7 +50,7 @@ def dao_get_uploads_by_service_id(service_id, limit_days=None, page=1, page_size
Job.service_id == service_id,
Job.original_file_name != current_app.config["TEST_MESSAGE_FILENAME"],
Job.original_file_name != current_app.config["ONE_OFF_MESSAGE_FILENAME"],
Job.job_status.notin_([JOB_STATUS_CANCELLED, JOB_STATUS_SCHEDULED]),
Job.job_status.notin_([JobStatus.CANCELLED, JobStatus.SCHEDULED]),
func.coalesce(Job.processing_started, Job.created_at)
>= today - func.coalesce(ServiceDataRetention.days_of_retention, 7),
]

2
app/enums.py
View File

@@ -178,7 +178,7 @@ class KeyType(Enum):
TEST = "test"
class JobStatusType(Enum):
class JobStatus(Enum):
PENDING = "pending"
IN_PROGRESS = "in progress"
FINISHED = "finished"

8
app/job/rest.py
View File

@@ -22,7 +22,7 @@ from app.dao.notifications_dao import (
from app.dao.services_dao import dao_fetch_service_by_id
from app.dao.templates_dao import dao_get_template_by_id
from app.errors import InvalidRequest, register_errors
from app.models import JOB_STATUS_CANCELLED, JOB_STATUS_PENDING, JOB_STATUS_SCHEDULED
from app.enums import JobStatus
from app.schemas import (
job_schema,
notification_with_template_schema,
@@ -53,7 +53,7 @@ def get_job_by_service_and_job_id(service_id, job_id):
@job_blueprint.route("/<job_id>/cancel", methods=["POST"])
def cancel_job(service_id, job_id):
job = dao_get_future_scheduled_job_by_id_and_service_id(job_id, service_id)
job.job_status = JOB_STATUS_CANCELLED
job.job_status = JobStatus.CANCELLED
dao_update_job(job)
return get_job_by_service_and_job_id(service_id, job_id)
@@ -175,13 +175,13 @@ def create_job(service_id):
job = job_schema.load(data)
if job.scheduled_for:
job.job_status = JOB_STATUS_SCHEDULED
job.job_status = JobStatus.SCHEDULED
dao_create_job(job)
sender_id = data.get("sender_id")
if job.job_status == JOB_STATUS_PENDING:
if job.job_status == JobStatus.PENDING:
process_job.apply_async(
[str(job.id)], {"sender_id": sender_id}, queue=QueueNames.JOBS
)

31
app/models.py
View File

@@ -28,6 +28,7 @@ from app.enums import (
CallbackType,
CodeType,
InvitedUserStatus,
JobStatus,
KeyType,
NotificationStatus,
NotificationType,
@@ -1346,33 +1347,6 @@ class ProviderDetailsHistory(db.Model, HistoryModel):
supports_international = db.Column(db.Boolean, nullable=False, default=False)
JOB_STATUS_PENDING = "pending"
JOB_STATUS_IN_PROGRESS = "in progress"
JOB_STATUS_FINISHED = "finished"
JOB_STATUS_SENDING_LIMITS_EXCEEDED = "sending limits exceeded"
JOB_STATUS_SCHEDULED = "scheduled"
JOB_STATUS_CANCELLED = "cancelled"
JOB_STATUS_READY_TO_SEND = "ready to send"
JOB_STATUS_SENT_TO_DVLA = "sent to dvla"
JOB_STATUS_ERROR = "error"
JOB_STATUS_TYPES = [
JOB_STATUS_PENDING,
JOB_STATUS_IN_PROGRESS,
JOB_STATUS_FINISHED,
JOB_STATUS_SENDING_LIMITS_EXCEEDED,
JOB_STATUS_SCHEDULED,
JOB_STATUS_CANCELLED,
JOB_STATUS_READY_TO_SEND,
JOB_STATUS_SENT_TO_DVLA,
JOB_STATUS_ERROR,
]
class JobStatus(db.Model):
__tablename__ = "job_status"
name = db.Column(db.String(255), primary_key=True)
class Job(db.Model):
__tablename__ = "jobs"
@@ -1423,8 +1397,7 @@ class Job(db.Model):
)
scheduled_for = db.Column(db.DateTime, index=True, unique=False, nullable=True)
job_status = db.Column(
db.String(255),
db.ForeignKey("job_status.name"),
db.Enum(JobStatus, name="job_status"),
index=True,
nullable=False,
default="pending",