isort .

app/enums.py

@@ -0,0 +1,69 @@
+from enum import StrEnum
+
+
+class NotificationStatus(StrEnum):
+    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(StrEnum):
+    NORMAL = "normal"
+    TEAM = "team"
+    TEST = "test"
+
+
+class JobStatus(StrEnum):
+    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(StrEnum):
+    ACCEPTED = "accepted"
+    CANCELLED = "cancelled"
+    EXPIRED = "expired"
+
+
+class InvitedOrgUserStatus(StrEnum):
+    ACCEPTED = "accepted"
+    CANCELLED = "cancelled"
+
+
+class VerificationStatus(StrEnum):
+    PENDING = "pending"
+    SUCCESS = "success"
+
+
+class AuthType(StrEnum):
+    EMAIL_AUTH = "email_auth"
+    SMS_AUTH = "sms_auth"
+
+
+# TODO:
+# class UserRole(StrEnum):
+#     ADMIN = "admin"
+#     USER = "user"
+#     GUEST = "guest"
+
+
+# TODO:
+# class NotificationType(StrEnum):
+#     EMAIL = "email"
+#     SMS = "sms"
+#     PUSH = "push"

app/main/views/activity.py

@@ -1,6 +1,7 @@
 from flask import abort, render_template, request, url_for
 
 from app import current_service, job_api_client
+from app.enums import NotificationStatus
 from app.formatters import get_time_left
 from app.main import main
 from app.utils.pagination import (
@@ -78,10 +79,6 @@ 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:
@@ -109,8 +106,8 @@ def create_job_dict_entry(job):
         "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),
+        "delivered_count": get_job_statistics(job, NotificationStatus.DELIVERED),
+        "failed_count": get_job_statistics(job, NotificationStatus.FAILED),
     }
 
 

app/main/views/invites.py

@@ -2,6 +2,7 @@ from flask import abort, flash, redirect, render_template, session, url_for
 from flask_login import current_user
 from markupsafe import Markup
 
+from app.enums import InvitedOrgUserStatus, InvitedUserStatus
 from app.main import main
 from app.models.organization import Organization
 from app.models.service import Service
@@ -29,14 +30,14 @@ def accept_invite(token):
 
         abort(403)
 
-    if invited_user.status == "cancelled":
+    if invited_user.status == InvitedUserStatus.CANCELLED:
         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:
         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:
         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:
         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 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:
         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:
         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

@@ -21,6 +21,7 @@ from app import (
     organizations_client,
     service_api_client,
 )
+from app.enums import VerificationStatus
 from app.event_handlers import (
     create_archive_service_event,
     create_resume_service_event,
@@ -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
     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
         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 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
 
     @property
     def scheduled(self):
-        return self.status == "scheduled"
+        return self.status == JobStatus.SCHEDULED
 
     @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, NotificationStatus.SENT
+        )
 
     @property
     def notifications_failed(self):
         return self._aggregate_statistics(
-            "failed",
-            "technical-failure",
-            "temporary-failure",
-            "permanent-failure",
-            "cancelled",
+            NotificationStatus.FAILED,
+            NotificationStatus.TECHNICAL_FAILURE,
+            NotificationStatus.TEMPORARY_FAILURE,
+            NotificationStatus.PERMANENT_FAILURE,
+            NotificationStatus.CANCELLED,
         )
 
     @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 or self.percentage_complete < 100
 
     @cached_property
     def finished_processing(self):

app/models/user.py

@@ -5,6 +5,7 @@ from flask import abort, current_app, request, session
 from flask_login import AnonymousUserMixin, UserMixin, login_user, logout_user
 from werkzeug.utils import cached_property
 
+from app.enums import InvitedUserStatus
 from app.event_handlers import (
     create_add_user_to_service_event,
     create_set_user_permissions_event,
@@ -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:
             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:
             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.enums import ApiKeyType
 from app.notify_client import NotifyAdminAPIClient, _attach_current_user
 
 # must match key types in notifications-api/app/models.py
-KEY_TYPE_NORMAL = "normal"
-KEY_TYPE_TEAM = "team"
-KEY_TYPE_TEST = "test"
+KEY_TYPE_NORMAL = ApiKeyType.NORMAL
+KEY_TYPE_TEAM = ApiKeyType.TEAM
+KEY_TYPE_TEST = ApiKeyType.TEST
 
 
 class ApiKeyApiClient(NotifyAdminAPIClient):

app/notify_client/invite_api_client.py

@@ -5,6 +5,7 @@ from urllib.parse import unquote
 from flask import current_app, request
 
 from app import redis_client
+from app.enums import InvitedUserStatus
 from app.notify_client import NotifyAdminAPIClient, _attach_current_user, cache
 from app.utils.user_permissions import (
     all_ui_permissions,
@@ -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}
         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}
         self.post(url=f"/service/{service_id}/invite/{invited_user_id}", data=data)
 
 

app/notify_client/job_api_client.py

@@ -1,6 +1,7 @@
 import datetime
 from zoneinfo import ZoneInfo
 
+from app.enums import JobStatus
 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
@@ -8,17 +9,17 @@ from app.utils.csv import get_user_preferred_timezone
 
 class JobApiClient(NotifyAdminAPIClient):
     JOB_STATUSES = {
-        "scheduled",
-        "pending",
-        "in progress",
-        "finished",
-        "cancelled",
-        "sending limits exceeded",
-        "ready to send",
-        "sent to dvla",
+        JobStatus.SCHEDULED,
+        JobStatus.PENDING,
+        JobStatus.IN_PROGRESS,
+        JobStatus.FINISHED,
+        JobStatus.CANCELLED,
+        JobStatus.SENDING_LIMITS_EXCEEDED,
+        JobStatus.READY_TO_SEND,
+        JobStatus.SENT_TO_DVLA,
     }
-    SCHEDULED_JOB_STATUS = "scheduled"
-    CANCELLED_JOB_STATUS = "cancelled"
+    SCHEDULED_JOB_STATUS = JobStatus.SCHEDULED
+    CANCELLED_JOB_STATUS = JobStatus.CANCELLED
     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
         )
 
     def get_page_of_jobs(self, service_id, *, page, statuses=None, limit_days=None):

app/notify_client/org_invite_api_client.py

@@ -1,3 +1,4 @@
+from app.enums import InvitedOrgUserStatus
 from app.notify_client import NotifyAdminAPIClient, _attach_current_user
 
 
@@ -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}
         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}
         self.post(
             url="/organization/{0}/invite/{1}".format(org_id, invited_user_id),
             data=data,

app/utils/__init__.py

@@ -7,16 +7,21 @@ from ordered_set import OrderedSet
 from werkzeug.datastructures import MultiDict
 from werkzeug.routing import RequestRedirect
 
+from app.enums import NotificationStatus
 from notifications_utils.field import Field
 
-SENDING_STATUSES = ["created", "pending", "sending"]
-DELIVERED_STATUSES = ["delivered", "sent"]
+SENDING_STATUSES = [
+    NotificationStatus.CREATED,
+    NotificationStatus.PENDING,
+    NotificationStatus.SENDING,
+]
+DELIVERED_STATUSES = [NotificationStatus.DELIVERED, NotificationStatus.SENT]
 FAILURE_STATUSES = [
-    "failed",
-    "temporary-failure",
-    "permanent-failure",
-    "technical-failure",
-    "validation-failed",
+    NotificationStatus.FAILED,
+    NotificationStatus.TEMPORARY_FAILURE,
+    NotificationStatus.PERMANENT_FAILURE,
+    NotificationStatus.TECHNICAL_FAILURE,
+    NotificationStatus.VALIDATION_FAILED,
 ]
 REQUESTED_STATUSES = SENDING_STATUSES + DELIVERED_STATUSES + FAILURE_STATUSES