enums part 2

This commit is contained in:
Beverly Nguyen
2025-07-11 14:18:54 -07:00
parent 0267a82687
commit e9f6b93d85
7 changed files with 141 additions and 84 deletions

View File

@@ -16,6 +16,33 @@ class NotificationStatus(StrEnum):
VALIDATION_FAILED = "validation-failed" VALIDATION_FAILED = "validation-failed"
CANCELLED = "cancelled" CANCELLED = "cancelled"
@classmethod
def sending_statuses(cls):
return [cls.CREATED, cls.PENDING, cls.SENDING]
@classmethod
def delivered_statuses(cls):
return [cls.DELIVERED, cls.SENT]
@classmethod
def failure_statuses(cls):
return [
cls.FAILED,
cls.TEMPORARY_FAILURE,
cls.PERMANENT_FAILURE,
cls.TECHNICAL_FAILURE,
cls.VALIDATION_FAILED,
]
@classmethod
def requested_statuses(cls):
return cls.sending_statuses() + cls.delivered_statuses() + cls.failure_statuses()
class NotificationType(StrEnum):
EMAIL = "email"
SMS = "sms"
class ApiKeyType(StrEnum): class ApiKeyType(StrEnum):
NORMAL = "normal" NORMAL = "normal"
@@ -60,10 +87,3 @@ class AuthType(StrEnum):
# ADMIN = "admin" # ADMIN = "admin"
# USER = "user" # USER = "user"
# GUEST = "guest" # GUEST = "guest"
# TODO:
# class NotificationType(StrEnum):
# EMAIL = "email"
# SMS = "sms"
# PUSH = "push"

View File

@@ -17,6 +17,7 @@ from flask import render_template_string, url_for
from flask.helpers import get_root_path from flask.helpers import get_root_path
from markupsafe import Markup from markupsafe import Markup
from app.enums import AuthType, NotificationStatus
from app.utils.csv import get_user_preferred_timezone from app.utils.csv import get_user_preferred_timezone
from app.utils.time import parse_naive_dt from app.utils.time import parse_naive_dt
from notifications_utils.field import Field from notifications_utils.field import Field
@@ -276,46 +277,53 @@ def format_notification_type(notification_type):
def format_notification_status(status, template_type): def format_notification_status(status, template_type):
return { return {
"email": { "email": {
"failed": "Failed", NotificationStatus.FAILED: "Failed",
"technical-failure": "Technical failure", NotificationStatus.TECHNICAL_FAILURE: "Technical failure",
"temporary-failure": "Inbox not accepting messages right now", NotificationStatus.TEMPORARY_FAILURE: "Inbox not accepting messages right now",
"permanent-failure": "Email address does not exist", NotificationStatus.PERMANENT_FAILURE: "Email address does not exist",
"delivered": "Delivered", NotificationStatus.DELIVERED: "Delivered",
"sending": "Sending", NotificationStatus.SENDING: "Sending",
"created": "Sending", NotificationStatus.CREATED: "Sending",
"sent": "Delivered", NotificationStatus.SENT: "Delivered",
}, },
"sms": { "sms": {
"failed": "Failed", NotificationStatus.FAILED: "Failed",
"technical-failure": "Technical failure", NotificationStatus.TECHNICAL_FAILURE: "Technical failure",
"temporary-failure": "Phone not accepting messages right now", NotificationStatus.TEMPORARY_FAILURE: "Phone not accepting messages right now",
"permanent-failure": "Not delivered", NotificationStatus.PERMANENT_FAILURE: "Not delivered",
"delivered": "Delivered", NotificationStatus.DELIVERED: "Delivered",
"sending": "Sending", NotificationStatus.SENDING: "Sending",
"created": "Sending", NotificationStatus.CREATED: "Sending",
"pending": "Sending", NotificationStatus.PENDING: "Sending",
"sent": "Sent", NotificationStatus.SENT: "Sent",
}, },
}[template_type].get(status, status) }[template_type].get(status, status)
def format_notification_status_as_time(status, created, updated): def format_notification_status_as_time(status, created, updated):
return dict.fromkeys( return dict.fromkeys(
{"created", "pending", "sending"}, " since {}".format(created) {
NotificationStatus.CREATED,
NotificationStatus.PENDING,
NotificationStatus.SENDING,
},
" since {}".format(created),
).get(status, updated) ).get(status, updated)
def format_notification_status_as_field_status(status, notification_type): def format_notification_status_as_field_status(status, notification_type):
return { return {
"failed": "error", NotificationStatus.FAILED: "error",
"technical-failure": "error", NotificationStatus.TECHNICAL_FAILURE: "error",
"temporary-failure": "error", NotificationStatus.TEMPORARY_FAILURE: "error",
"permanent-failure": "error", NotificationStatus.PERMANENT_FAILURE: "error",
"delivered": None, NotificationStatus.DELIVERED: None,
"sent": "sent-international" if notification_type == "sms" else None, NotificationStatus.SENT: (
"sending": "default", "sent-international" if notification_type == "sms" else None
"created": "default", ),
"pending": "default", NotificationStatus.SENDING: "default",
NotificationStatus.CREATED: "default",
NotificationStatus.PENDING: "default",
}.get(status, "error") }.get(status, "error")
@@ -323,9 +331,9 @@ def format_notification_status_as_url(status, notification_type):
url = partial(url_for, "main.message_status") url = partial(url_for, "main.message_status")
if status not in { if status not in {
"technical-failure", NotificationStatus.TECHNICAL_FAILURE,
"temporary-failure", NotificationStatus.TEMPORARY_FAILURE,
"permanent-failure", NotificationStatus.PERMANENT_FAILURE,
}: }:
return None return None
@@ -559,8 +567,8 @@ def square_metres_to_square_miles(area):
def format_auth_type(auth_type, with_indefinite_article=False): def format_auth_type(auth_type, with_indefinite_article=False):
indefinite_article, auth_type = { indefinite_article, auth_type = {
"email_auth": ("an", "Email link"), AuthType.EMAIL_AUTH: ("an", "Email link"),
"sms_auth": ("a", "Text message code"), AuthType.SMS_AUTH: ("a", "Text message code"),
}[auth_type] }[auth_type]
if with_indefinite_article: if with_indefinite_article:

View File

@@ -14,6 +14,7 @@ from app import (
service_api_client, service_api_client,
template_statistics_client, template_statistics_client,
) )
from app.enums import JobStatus, NotificationStatus
from app.main import main from app.main import main
from app.main.views.user_profile import set_timezone from app.main.views.user_profile import set_timezone
from app.statistics_utils import get_formatted_percentage from app.statistics_utils import get_formatted_percentage
@@ -42,7 +43,9 @@ def service_dashboard(service_id):
job_response = job_api_client.get_jobs(service_id)["data"] job_response = job_api_client.get_jobs(service_id)["data"]
service_data_retention_days = 7 service_data_retention_days = 7
active_jobs = [job for job in job_response if job["job_status"] != "cancelled"] active_jobs = [
job for job in job_response if job["job_status"] != JobStatus.CANCELLED
]
job_lists = [ job_lists = [
{**job_dict, "finished_processing": job_is_finished(job_dict)} {**job_dict, "finished_processing": job_is_finished(job_dict)}
for job_dict in active_jobs for job_dict in active_jobs
@@ -69,7 +72,9 @@ def service_dashboard(service_id):
def job_is_finished(job_dict): def job_is_finished(job_dict):
done_statuses = DELIVERED_STATUSES + FAILURE_STATUSES + ["cancelled"] done_statuses = (
DELIVERED_STATUSES + FAILURE_STATUSES + [NotificationStatus.CANCELLED]
)
processed_count = sum( processed_count = sum(
stat["count"] stat["count"]
for stat in job_dict["statistics"] for stat in job_dict["statistics"]
@@ -106,8 +111,18 @@ def get_local_daily_stats_for_last_x_days(stats_utc, user_timezone, days):
] ]
aggregator = { aggregator = {
d: { d: {
"sms": {"delivered": 0, "failure": 0, "pending": 0, "requested": 0}, "sms": {
"email": {"delivered": 0, "failure": 0, "pending": 0, "requested": 0}, NotificationStatus.DELIVERED: 0,
"failure": 0,
NotificationStatus.PENDING: 0,
"requested": 0,
},
"email": {
NotificationStatus.DELIVERED: 0,
"failure": 0,
NotificationStatus.PENDING: 0,
"requested": 0,
},
} }
for d in days_list for d in days_list
} }
@@ -121,7 +136,12 @@ def get_local_daily_stats_for_last_x_days(stats_utc, user_timezone, days):
if local_day in aggregator: if local_day in aggregator:
for msg_type in ["sms", "email"]: for msg_type in ["sms", "email"]:
for status in ["delivered", "failure", "pending", "requested"]: for status in [
NotificationStatus.DELIVERED,
"failure",
NotificationStatus.PENDING,
"requested",
]:
aggregator[local_day][msg_type][status] += data[msg_type][status] aggregator[local_day][msg_type][status] += data[msg_type][status]
return aggregator return aggregator
@@ -231,7 +251,9 @@ def usage(service_id):
def filter_out_cancelled_stats(template_statistics): def filter_out_cancelled_stats(template_statistics):
return [s for s in template_statistics if s["status"] != "cancelled"] return [
s for s in template_statistics if s["status"] != NotificationStatus.CANCELLED
]
def aggregate_template_usage(template_statistics, sort_key="count"): def aggregate_template_usage(template_statistics, sort_key="count"):
@@ -265,7 +287,7 @@ def get_dashboard_totals(statistics):
for msg_type in statistics.values(): for msg_type in statistics.values():
msg_type["failed_percentage"] = get_formatted_percentage( msg_type["failed_percentage"] = get_formatted_percentage(
msg_type["failed"], msg_type["requested"] msg_type[NotificationStatus.FAILED], msg_type["requested"]
) )
msg_type["show_warning"] = float(msg_type["failed_percentage"]) > 3 msg_type["show_warning"] = float(msg_type["failed_percentage"]) > 3
@@ -314,7 +336,9 @@ def aggregate_status_types(counts_dict):
return get_dashboard_totals( return get_dashboard_totals(
{ {
"{}_counts".format(message_type): { "{}_counts".format(message_type): {
"failed": sum(stats.get(status, 0) for status in FAILURE_STATUSES), NotificationStatus.FAILED: sum(
stats.get(status, 0) for status in FAILURE_STATUSES
),
"requested": sum(stats.get(status, 0) for status in REQUESTED_STATUSES), "requested": sum(stats.get(status, 0) for status in REQUESTED_STATUSES),
} }
for message_type, stats in counts_dict.items() for message_type, stats in counts_dict.items()

View File

@@ -23,7 +23,7 @@ from app import (
notification_api_client, notification_api_client,
service_api_client, service_api_client,
) )
from app.enums import JobStatus from app.enums import JobStatus, NotificationStatus
from app.formatters import get_time_left, message_count_noun from app.formatters import get_time_left, message_count_noun
from app.main import main from app.main import main
from app.main.forms import SearchNotificationsForm from app.main.forms import SearchNotificationsForm
@@ -304,22 +304,30 @@ def get_status_filters(service, message_type, statistics):
if message_type is None: if message_type is None:
stats = { stats = {
key: sum(statistics[message_type][key] for message_type in {"email", "sms"}) key: sum(statistics[message_type][key] for message_type in {"email", "sms"})
for key in {"requested", "delivered", "failed"} for key in {
"requested",
NotificationStatus.DELIVERED,
NotificationStatus.FAILED,
}
} }
else: else:
stats = statistics[message_type] stats = statistics[message_type]
if stats.get("failure") is not None: if stats.get("failure") is not None:
stats["failed"] = stats["failure"] stats[NotificationStatus.FAILED] = stats["failure"]
stats["pending"] = stats["requested"] - stats["delivered"] - stats["failed"] stats[NotificationStatus.PENDING] = (
stats["requested"]
- stats[NotificationStatus.DELIVERED]
- stats[NotificationStatus.FAILED]
)
filters = [ filters = [
# key, label, option # key, label, option
("requested", "total", "sending,delivered,failed"), ("requested", "total", "sending,delivered,failed"),
("pending", "pending", "sending,pending"), (NotificationStatus.PENDING, "pending", "sending,pending"),
("delivered", "delivered", "delivered"), (NotificationStatus.DELIVERED, "delivered", "delivered"),
("failed", "failed", "failed"), (NotificationStatus.FAILED, "failed", "failed"),
] ]
return [ return [
# return list containing label, option, link, count # return list containing label, option, link, count

View File

@@ -25,6 +25,7 @@ from app import (
service_api_client, service_api_client,
user_api_client, user_api_client,
) )
from app.enums import NotificationStatus
from app.extensions import redis_client from app.extensions import redis_client
from app.main import main from app.main import main
from app.main.forms import ( from app.main.forms import (
@@ -769,32 +770,41 @@ def filter_and_sort_services(services, trial_mode_services=False):
def create_global_stats(services): def create_global_stats(services):
stats = { stats = {
"email": {"delivered": 0, "failed": 0, "requested": 0}, "email": {
"sms": {"delivered": 0, "failed": 0, "requested": 0}, NotificationStatus.DELIVERED: 0,
NotificationStatus.FAILED: 0,
"requested": 0,
},
"sms": {
NotificationStatus.DELIVERED: 0,
NotificationStatus.FAILED: 0,
"requested": 0,
},
} }
# Issue #1323. The back end is now sending 'failure' instead of # Issue #1323. The back end is now sending 'failure' instead of
# 'failed'. Adjust it here, but keep it flexible in case # 'failed'. Adjust it here, but keep it flexible in case
# the backend reverts to 'failed'. # the backend reverts to 'failed'.
for service in services: for service in services:
if service["statistics"]["sms"].get("failure") is not None: if service["statistics"]["sms"].get("failure") is not None:
service["statistics"]["sms"]["failed"] = service["statistics"]["sms"][ service["statistics"]["sms"][NotificationStatus.FAILED] = service[
"failure" "statistics"
] ]["sms"]["failure"]
if service["statistics"]["email"].get("failure") is not None: if service["statistics"]["email"].get("failure") is not None:
service["statistics"]["email"]["failed"] = service["statistics"]["email"][ service["statistics"]["email"][NotificationStatus.FAILED] = service[
"failure" "statistics"
] ]["email"]["failure"]
for service in services: for service in services:
for msg_type, status in itertools.product( for msg_type, status in itertools.product(
("sms", "email"), ("delivered", "failed", "requested") ("sms", "email"),
(NotificationStatus.DELIVERED, NotificationStatus.FAILED, "requested"),
): ):
stats[msg_type][status] += service["statistics"][msg_type][status] stats[msg_type][status] += service["statistics"][msg_type][status]
for stat in stats.values(): for stat in stats.values():
stat["failure_rate"] = get_formatted_percentage( stat["failure_rate"] = get_formatted_percentage(
stat["failed"], stat["requested"] stat[NotificationStatus.FAILED], stat["requested"]
) )
return stats return stats

View File

@@ -7,25 +7,15 @@ from ordered_set import OrderedSet
from werkzeug.datastructures import MultiDict from werkzeug.datastructures import MultiDict
from werkzeug.routing import RequestRedirect from werkzeug.routing import RequestRedirect
from app.enums import NotificationStatus from app.enums import NotificationStatus, NotificationType
from notifications_utils.field import Field from notifications_utils.field import Field
SENDING_STATUSES = [ SENDING_STATUSES = NotificationStatus.sending_statuses()
NotificationStatus.CREATED, DELIVERED_STATUSES = NotificationStatus.delivered_statuses()
NotificationStatus.PENDING, FAILURE_STATUSES = NotificationStatus.failure_statuses()
NotificationStatus.SENDING, REQUESTED_STATUSES = NotificationStatus.requested_statuses()
]
DELIVERED_STATUSES = [NotificationStatus.DELIVERED, NotificationStatus.SENT]
FAILURE_STATUSES = [
NotificationStatus.FAILED,
NotificationStatus.TEMPORARY_FAILURE,
NotificationStatus.PERMANENT_FAILURE,
NotificationStatus.TECHNICAL_FAILURE,
NotificationStatus.VALIDATION_FAILED,
]
REQUESTED_STATUSES = SENDING_STATUSES + DELIVERED_STATUSES + FAILURE_STATUSES
NOTIFICATION_TYPES = ["sms", "email"] NOTIFICATION_TYPES = [NotificationType.SMS, NotificationType.EMAIL]
def service_has_permission(permission): def service_has_permission(permission):

View File

@@ -63,10 +63,7 @@ def test_organization_page_shows_all_organizations(
assert normalize_spaces(archived.text) == "- archived" assert normalize_spaces(archived.text) == "- archived"
assert normalize_spaces(archived.parent.text) == "Test 2 - archived 2 live services" assert normalize_spaces(archived.parent.text) == "Test 2 - archived 2 live services"
assert ( assert normalize_spaces(page.select_one("a.usa-button").text) == "New organization"
normalize_spaces(page.select_one("a.usa-button").text)
== "New organization"
)
get_organizations.assert_called_once_with() get_organizations.assert_called_once_with()