darkhelm/notifications-admin
1
0
Fork 0
mirror of https://github.com/GSA/notifications-admin.git synced 2026-08-20 14:29:51 -04:00
Code Issues Packages Projects Releases Wiki Activity

Improved job/activity list ordering and display. Jobs are now shown by most recent activity (processing started or created)

Browse Source
This commit is contained in:
Beverly Nguyen
2025-06-25 13:25:01 -07:00
parent 543e8f2f44
commit 5007523d79
3 changed files with 46 additions and 47 deletions
Download Patch File Download Diff File Expand all files Collapse all files

84
app/main/views/activity.py
View File

@@ -78,47 +78,45 @@ def handle_pagination(jobs, service_id, page):
return prev_page, next_page, pagination
JOB_STATUS_DELIVERED = "delivered"
JOB_STATUS_FAILED = "failed"
def get_job_statistics(job, status):
statistics = job.get("statistics", [])
for stat in statistics:
if stat.get("status") == status:
return stat.get("count")
return None
def create_job_dict_entry(job):
job_id = job.get("id")
can_download = get_time_left(job.get("created_at")) != "Data no longer available"
activity_time = job.get("processing_started") or job.get("created_at")
return {
"job_id": job_id,
"can_download": can_download,
"download_link": url_for(
".view_job_csv",
service_id=current_service.id,
job_id=job_id
) if can_download else None,
"view_job_link": url_for(
".view_job",
service_id=current_service.id,
job_id=job_id
),
"activity_time": activity_time,
"created_by": job.get("created_by"),
"template_name": job.get("template_name"),
"delivered_count": get_job_statistics(job, JOB_STATUS_DELIVERED),
"failed_count": get_job_statistics(job, JOB_STATUS_FAILED),
}
def generate_job_dict(jobs):
return [
{
"job_id": job["id"],
"time_left": get_time_left(job["created_at"]),
"download_link": url_for(
".view_job_csv", service_id=current_service.id, job_id=job["id"]
),
"view_job_link": url_for(
".view_job", service_id=current_service.id, job_id=job["id"]
),
"created_at": job["created_at"],
"time_sent_data_value": convert_time_unixtimestamp(
job["processing_finished"]
if job["processing_finished"]
else (
job["processing_started"]
if job["processing_started"]
else job["created_at"]
)
),
"processing_finished": job["processing_finished"],
"processing_started": job["processing_started"],
"created_by": job["created_by"],
"template_name": job["template_name"],
"delivered_count": next(
(
stat["count"]
for stat in job.get("statistics", [])
if stat["status"] == "delivered"
),
None,
),
"failed_count": next(
(
stat["count"]
for stat in job.get("statistics", [])
if stat["status"] == "failed"
),
None,
),
}
for job in jobs["data"]
]
if not jobs or not jobs.get("data"):
return []
return [create_job_dict_entry(job) for job in jobs["data"]]