PART 1: Created and implemented enums throughout the codebase to replace hardcoded status strings, improving type safety and reducing the risk of typos.

app/enums.py

@@ -0,0 +1,74 @@
+from enum import Enum
+
+
+class NotificationStatus(Enum):
+    CREATED = "created"
+    PENDING = "pending"
+    SENDING = "sending"
+
+    DELIVERED = "delivered"
+    SENT = "sent"
+
+    FAILED = "failed"
+    TEMPORARY_FAILURE = "temporary-failure"
+    PERMANENT_FAILURE = "permanent-failure"
+    TECHNICAL_FAILURE = "technical-failure"
+    VALIDATION_FAILED = "validation-failed"
+    CANCELLED = "cancelled"
+
+
+class ApiKeyType(Enum):
+    NORMAL = "normal"
+    TEAM = "team"
+    TEST = "test"
+
+
+class JobStatus(Enum):
+    PENDING = "pending"
+    IN_PROGRESS = "in progress"
+    FINISHED = "finished"
+    SENDING_LIMITS_EXCEEDED = "sending limits exceeded"
+    SCHEDULED = "scheduled"
+    CANCELLED = "cancelled"
+    READY_TO_SEND = "ready to send"
+    SENT_TO_DVLA = "sent to dvla"
+
+
+class InvitedUserStatus(Enum):
+    ACCEPTED = "accepted"
+    CANCELLED = "cancelled"
+    EXPIRED = "expired"
+
+
+class InvitedOrgUserStatus(Enum):
+    ACCEPTED = "accepted"
+    CANCELLED = "cancelled"
+
+
+class VerificationStatus(Enum):
+    PENDING = "pending"
+    SUCCESS = "success"
+
+
+class HealthStatus(Enum):
+    OK = "ok"
+    ERROR = "error"
+
+
+class AuthType(Enum):
+    EMAIL_AUTH = "email_auth"
+    SMS_AUTH = "sms_auth"
+
+
+# TODO: UserRole enum
+# class UserRole(Enum):
+#     ADMIN = "admin"
+#     USER = "user"
+#     GUEST = "guest"
+
+
+# TODO: NotificationType enum
+# class NotificationType(Enum):
+#     EMAIL = "email"
+#     SMS = "sms"
+#     PUSH = "push"

app/main/views/invites.py

@@ -6,6 +6,7 @@ from app.main import main
 from app.models.organization import Organization
 from app.models.service import Service
 from app.models.user import InvitedOrgUser, InvitedUser, OrganizationUsers, User, Users
+from app.enums import InvitedUserStatus, InvitedOrgUserStatus
 
 
 @main.route("/invitation/<token>")
@@ -29,14 +30,14 @@ def accept_invite(token):
 
         abort(403)
 
-    if invited_user.status == "cancelled":
+    if invited_user.status == InvitedUserStatus.CANCELLED.value:
         service = Service.from_id(invited_user.service)
         return render_template(
             "views/cancelled-invitation.html",
             from_user=invited_user.from_user.name,
             service_name=service.name,
         )
-    if invited_user.status == "accepted":
+    if invited_user.status == InvitedUserStatus.ACCEPTED.value:
         session.pop("invited_user_id", None)
         service = Service.from_id(invited_user.service)
         return redirect(
@@ -104,7 +105,7 @@ def accept_org_invite(token):
 
         abort(403)
 
-    if invited_org_user.status == "cancelled":
+    if invited_org_user.status == InvitedOrgUserStatus.CANCELLED.value:
         organization = Organization.from_id(invited_org_user.organization)
         return render_template(
             "views/cancelled-invitation.html",
@@ -112,7 +113,7 @@ def accept_org_invite(token):
             organization_name=organization.name,
         )
 
-    if invited_org_user.status == "accepted":
+    if invited_org_user.status == InvitedOrgUserStatus.ACCEPTED.value:
         session.pop("invited_org_user_id", None)
         return redirect(
             url_for("main.organization_dashboard", org_id=invited_org_user.organization)

app/main/views/jobs.py

@@ -23,6 +23,7 @@ from app import (
     notification_api_client,
     service_api_client,
 )
+from app.enums import JobStatus
 from app.formatters import get_time_left, message_count_noun
 from app.main import main
 from app.main.forms import SearchNotificationsForm
@@ -398,7 +399,7 @@ def get_job_partials(job):
         counts=_get_job_counts(job),
         status=filter_args["status"],
         notifications_deleted=(
-            job.status == "finished" and not notifications["notifications"]
+            job.status == JobStatus.FINISHED.value and not notifications["notifications"]
         ),
     )
     service_data_retention_days = current_service.get_days_of_retention(

app/main/views/register.py

@@ -15,6 +15,7 @@ from flask import (
 )
 
 from app import redis_client, user_api_client
+from app.enums import InvitedUserStatus
 from app.main import main
 from app.main.forms import (
     RegisterUserFromOrgInviteForm,
@@ -254,14 +255,14 @@ def get_invited_user_email_address(invited_user_id):
 def invited_user_accept_invite(invited_user_id):
     invited_user = InvitedUser.by_id(invited_user_id)
 
-    if invited_user.status == "expired":
+    if invited_user.status == InvitedUserStatus.EXPIRED.value:
         current_app.logger.error("User invitation has expired")
         flash(
             "Your invitation has expired; please contact the person who invited you for additional help."
         )
         abort(401)
 
-    if invited_user.status == "cancelled":
+    if invited_user.status == InvitedUserStatus.CANCELLED.value:
         current_app.logger.error("User invitation has been cancelled")
         flash(
             "Your invitation is no longer valid; please contact the person who invited you for additional help."

app/main/views/service_settings.py

@@ -52,6 +52,7 @@ from app.utils import DELIVERED_STATUSES, FAILURE_STATUSES, SENDING_STATUSES
 from app.utils.time import parse_naive_dt
 from app.utils.user import user_has_permissions, user_is_platform_admin
 from notifications_python_client.errors import HTTPError
+from app.enums import VerificationStatus
 
 PLATFORM_ADMIN_SERVICE_PERMISSIONS = OrderedDict(
     [
@@ -397,10 +398,10 @@ def get_service_verify_reply_to_address_partials(service_id, notification_id):
     if replace:
         existing = current_service.get_email_reply_to_address(replace)
         existing_is_default = existing["is_default"]
-    verification_status = "pending"
+    verification_status = VerificationStatus.PENDING.value
     is_default = True if (request.args.get("is_default", False) == "True") else False
     if notification["status"] in DELIVERED_STATUSES:
-        verification_status = "success"
+        verification_status = VerificationStatus.SUCCESS.value
         if notification["to"] not in [
             i["email_address"] for i in current_service.email_reply_to_addresses
         ]:
@@ -441,7 +442,7 @@ def get_service_verify_reply_to_address_partials(service_id, notification_id):
             first_email_address=first_email_address,
             replace=replace,
         ),
-        "stop": 0 if verification_status == "pending" else 1,
+        "stop": 0 if verification_status == VerificationStatus.PENDING.value else 1,
     }
 
 

app/models/job.py

@@ -1,5 +1,6 @@
 from werkzeug.utils import cached_property
 
+from app.enums import JobStatus, NotificationStatus
 from app.models import JSONModel, ModelList, PaginatedModelList
 from app.notify_client.job_api_client import job_api_client
 from app.notify_client.notification_api_client import notification_api_client
@@ -32,11 +33,11 @@ class Job(JSONModel):
 
     @property
     def cancelled(self):
-        return self.status == "cancelled"
+        return self.status == JobStatus.CANCELLED.value
 
     @property
     def scheduled(self):
-        return self.status == "scheduled"
+        return self.status == JobStatus.SCHEDULED.value
 
     @property
     def scheduled_for(self):
@@ -61,16 +62,18 @@ class Job(JSONModel):
 
     @property
     def notifications_delivered(self):
-        return self._aggregate_statistics("delivered", "sent")
+        return self._aggregate_statistics(
+            NotificationStatus.DELIVERED.value, NotificationStatus.SENT.value
+        )
 
     @property
     def notifications_failed(self):
         return self._aggregate_statistics(
-            "failed",
+            NotificationStatus.FAILED.value,
-            "technical-failure",
+            NotificationStatus.TECHNICAL_FAILURE.value,
-            "temporary-failure",
+            NotificationStatus.TEMPORARY_FAILURE.value,
-            "permanent-failure",
+            NotificationStatus.PERMANENT_FAILURE.value,
-            "cancelled",
+            NotificationStatus.CANCELLED.value,
         )
 
     @property
@@ -95,7 +98,7 @@ class Job(JSONModel):
 
     @property
     def still_processing(self):
-        return self.status != "finished" or self.percentage_complete < 100
+        return self.status != JobStatus.FINISHED.value or self.percentage_complete < 100
 
     @cached_property
     def finished_processing(self):

app/models/user.py

@@ -22,6 +22,7 @@ from app.utils.user_permissions import (
     translate_permissions_from_db_to_ui,
 )
 from notifications_python_client.errors import HTTPError
+from app.enums import InvitedUserStatus
 
 
 def _get_service_id_from_view_args():
@@ -566,7 +567,7 @@ class InvitedUser(JSONModel):
         # current_app.logger.warning(
         #    f"Checking invited user {self.id} for permissions: {permissions}"
         # )
-        if self.status == "cancelled":
+        if self.status == InvitedUserStatus.CANCELLED.value:
             return False
         return set(self.permissions) > set(permissions)
 
@@ -574,7 +575,7 @@ class InvitedUser(JSONModel):
         # current_app.logger.warn(
         #    f"Checking invited user {self.id} for permission: {permission} on service {service_id}"
         # )
-        if self.status == "cancelled":
+        if self.status == InvitedUserStatus.CANCELLED.value:
             return False
         return self.service == service_id and permission in self.permissions
 

app/notify_client/api_key_api_client.py

@@ -1,9 +1,10 @@
 from app.notify_client import NotifyAdminAPIClient, _attach_current_user
+from app.enums import ApiKeyType
 
 # must match key types in notifications-api/app/models.py
-KEY_TYPE_NORMAL = "normal"
+KEY_TYPE_NORMAL = ApiKeyType.NORMAL.value
-KEY_TYPE_TEAM = "team"
+KEY_TYPE_TEAM = ApiKeyType.TEAM.value
-KEY_TYPE_TEST = "test"
+KEY_TYPE_TEST = ApiKeyType.TEST.value
 
 
 class ApiKeyApiClient(NotifyAdminAPIClient):

app/notify_client/invite_api_client.py

@@ -11,6 +11,7 @@ from app.utils.user_permissions import (
     translate_permissions_from_ui_to_db,
 )
 from notifications_utils.url_safe_token import generate_token
+from app.enums import InvitedUserStatus
 
 
 class InviteApiClient(NotifyAdminAPIClient):
@@ -94,7 +95,7 @@ class InviteApiClient(NotifyAdminAPIClient):
         return self.get(url=f"/invite/service/check/{token}")["data"]
 
     def cancel_invited_user(self, service_id, invited_user_id):
-        data = {"status": "cancelled"}
+        data = {"status": InvitedUserStatus.CANCELLED.value}
         data = _attach_current_user(data)
         self.post(url=f"/service/{service_id}/invite/{invited_user_id}", data=data)
 
@@ -131,7 +132,7 @@ class InviteApiClient(NotifyAdminAPIClient):
     @cache.delete("service-{service_id}")
     @cache.delete("user-{invited_user_id}")
     def accept_invite(self, service_id, invited_user_id):
-        data = {"status": "accepted"}
+        data = {"status": InvitedUserStatus.ACCEPTED.value}
         self.post(url=f"/service/{service_id}/invite/{invited_user_id}", data=data)
 
 

app/notify_client/job_api_client.py

@@ -4,21 +4,22 @@ from zoneinfo import ZoneInfo
 from app.extensions import redis_client
 from app.notify_client import NotifyAdminAPIClient, _attach_current_user, cache
 from app.utils.csv import get_user_preferred_timezone
+from app.enums import JobStatus
 
 
 class JobApiClient(NotifyAdminAPIClient):
     JOB_STATUSES = {
-        "scheduled",
+        JobStatus.SCHEDULED.value,
-        "pending",
+        JobStatus.PENDING.value,
-        "in progress",
+        JobStatus.IN_PROGRESS.value,
-        "finished",
+        JobStatus.FINISHED.value,
-        "cancelled",
+        JobStatus.CANCELLED.value,
-        "sending limits exceeded",
+        JobStatus.SENDING_LIMITS_EXCEEDED.value,
-        "ready to send",
+        JobStatus.READY_TO_SEND.value,
-        "sent to dvla",
+        JobStatus.SENT_TO_DVLA.value,
     }
-    SCHEDULED_JOB_STATUS = "scheduled"
+    SCHEDULED_JOB_STATUS = JobStatus.SCHEDULED.value
-    CANCELLED_JOB_STATUS = "cancelled"
+    CANCELLED_JOB_STATUS = JobStatus.CANCELLED.value
     NON_CANCELLED_JOB_STATUSES = JOB_STATUSES - {CANCELLED_JOB_STATUS}
     NON_SCHEDULED_JOB_STATUSES = JOB_STATUSES - {
         SCHEDULED_JOB_STATUS,
@@ -57,7 +58,7 @@ class JobApiClient(NotifyAdminAPIClient):
                 job["original_file_name"],
             )
             for job in self.get_jobs(service_id, limit_days=0)["data"]
-            if job["job_status"] != "cancelled"
+            if job["job_status"] != JobStatus.CANCELLED.value
         )
 
     def get_page_of_jobs(self, service_id, *, page, statuses=None, limit_days=None):

app/notify_client/org_invite_api_client.py

@@ -1,4 +1,5 @@
 from app.notify_client import NotifyAdminAPIClient, _attach_current_user
+from app.enums import InvitedOrgUserStatus
 
 
 class OrgInviteApiClient(NotifyAdminAPIClient):
@@ -33,7 +34,7 @@ class OrgInviteApiClient(NotifyAdminAPIClient):
         return resp["data"]
 
     def cancel_invited_user(self, org_id, invited_user_id):
-        data = {"status": "cancelled"}
+        data = {"status": InvitedOrgUserStatus.CANCELLED.value}
         data = _attach_current_user(data)
         self.post(
             url="/organization/{0}/invite/{1}".format(org_id, invited_user_id),
@@ -41,7 +42,7 @@ class OrgInviteApiClient(NotifyAdminAPIClient):
         )
 
     def accept_invite(self, org_id, invited_user_id):
-        data = {"status": "accepted"}
+        data = {"status": InvitedOrgUserStatus.ACCEPTED.value}
         self.post(
             url="/organization/{0}/invite/{1}".format(org_id, invited_user_id),
             data=data,

app/status/views/healthcheck.py

@@ -5,6 +5,7 @@ from flask import current_app, jsonify, request
 from redis import RedisError
 
 from app import status_api_client, version
+from app.enums import HealthStatus
 from app.extensions import redis_client
 from app.status import status
 from notifications_python_client.errors import HTTPError
@@ -13,16 +14,16 @@ from notifications_python_client.errors import HTTPError
 @status.route("/_status", methods=["GET"])
 def show_status():
     if request.args.get("elb", None) or request.args.get("simple", None):
-        return jsonify(status="ok"), 200
+        return jsonify(status=HealthStatus.OK.value), 200
     else:
         try:
             api_status = status_api_client.get_status()
         except HTTPError as err:
             current_app.logger.exception("API failed to respond")
-            return jsonify(status="error", message=str(err.message)), 500
+            return jsonify(status=HealthStatus.ERROR.value, message=str(err.message)), 500
         return (
             jsonify(
-                status="ok",
+                status=HealthStatus.OK.value,
                 api=api_status,
                 git_commit=version.__git_commit__,
                 build_time=version.__time__,
@@ -60,10 +61,10 @@ def show_redis_status():
             )
         except HTTPError as err:
             current_app.logger.exception("API failed to respond")
-            return jsonify(status="error", message=str(err.message)), 500
+            return jsonify(status=HealthStatus.ERROR.value, message=str(err.message)), 500
         return (
             jsonify(
-                status="ok",
+                status=HealthStatus.OK.value,
                 api=api_status,
                 git_commit=version.__git_commit__,
                 build_time=version.__time__,
@@ -76,7 +77,7 @@ def show_redis_status():
         )
         return (
             jsonify(
-                status=f"error: {err}",
+                status=f"{HealthStatus.ERROR.value}: {err}",
                 api=api_status,
                 git_commit=version.__git_commit__,
                 build_time=version.__time__,