move task status sets to a classmethod

src/fides/api/models/worker_task.py

@@ -1,5 +1,5 @@
 import enum
-from typing import Any, List
+from typing import Any, List, Set
 
 from sqlalchemy import Column, DateTime, String
 from sqlalchemy.sql import text
@@ -19,12 +19,31 @@ class ExecutionLogStatus(enum.Enum):
     skipped = "skipped"
     polling = "polling"
 
+    @classmethod
+    def in_progress_statuses(cls) -> Set["ExecutionLogStatus"]:
+        """
+        Statuses that indicate the task is in progress.
+
+        For now, used only to detect interrupted monitor tasks.
+        """
+        # TODO: update this to be more generic, i.e. used for other task types?
+        return {
+            cls.in_processing,
+            cls.pending,
+        }
 
-# Statuses that can be resumed
-RESUMABLE_EXECUTION_LOG_STATUSES = [
-    ExecutionLogStatus.pending,
-    ExecutionLogStatus.polling,
-]
+    @classmethod
+    def resumable_statuses(cls) -> Set["ExecutionLogStatus"]:
+        """
+        Statuses that can be resumed.
+
+        For now, used only for DSR tasks.
+        """
+        # TODO: update this to be more generic, i.e. used for other task types?
+        return {
+            cls.pending,
+            cls.polling,
+        }
 
 
 class WorkerTask:

src/fides/api/task/execute_request_tasks.py

@@ -23,10 +23,7 @@
 from fides.api.models.connectionconfig import ConnectionConfig
 from fides.api.models.policy import Policy
 from fides.api.models.privacy_request import ExecutionLog, PrivacyRequest, RequestTask
-from fides.api.models.worker_task import (
-    RESUMABLE_EXECUTION_LOG_STATUSES,
-    ExecutionLogStatus,
-)
+from fides.api.models.worker_task import ExecutionLogStatus
 from fides.api.schemas.policy import ActionType, CurrentStep
 from fides.api.schemas.privacy_request import PrivacyRequestStatus
 from fides.api.task.graph_task import (
@@ -178,7 +175,7 @@ def can_run_task_body(
     if request_task.is_root_task:
         # Shouldn't be possible but adding as a catch-all
         return False
-    if request_task.status not in RESUMABLE_EXECUTION_LOG_STATUSES:
+    if request_task.status not in ExecutionLogStatus.resumable_statuses():
         logger_method(request_task)(
             "Skipping {} task {} with status {}.",
             request_task.action_type,