Slack alert throttle

files/.gitignore

@@ -0,0 +1,8 @@
+# Virtual Environments
+venv/
+.venv/
+
+# Byte-compiled / optimized / DLL files
+__pycache__/
+*.py[cod]
+*$py.class

files/requirements.txt

@@ -1,3 +1,3 @@
-psycopg2-binary
-requests
-pyyaml
+psycopg2-binary>=2.9,<3.0
+PyYAML>=6.0,<7.0
+requests>=2.32,<3.0

files/test.py

@@ -25,8 +25,7 @@ def test_contains_new_entry(self):
 
     def test_contains_existing_entry(self):
         jwd = "existing_id"
-        with open(self.temp_file.name, "a") as f:
-            f.write(f"{datetime.now()}\t{jwd}\n")
+        self.record._write_record(jwd)
         self.assertTrue(self.record.contains(jwd), "Existing entry should return True")
 
     @patch("walle.SLACK_NOTIFY_PERIOD_DAYS", new=SLACK_NOTIFY_PERIOD_DAYS)

files/walle.py

@@ -16,6 +16,7 @@
 import sys
 import time
 import zlib
+from dataclasses import dataclass
 from datetime import datetime, timedelta
 from typing import Dict, List, Union
 
@@ -63,6 +64,20 @@ def convert_arg_to_seconds(hours: str) -> float:
     return float(hours) * 60 * 60
 
 
+@dataclass
+class Record:
+    date: str
+    jwd: str
+
+    def __post_init__(self):
+        if not (
+            isinstance(self.date, str) and isinstance(self.jwd, (str, pathlib.Path))
+        ):
+            raise ValueError
+        self.jwd = str(self.jwd)
+        datetime.fromisoformat(self.date)  # will raise ValueError if invalid
+
+
 class NotificationHistory:
     """Record of Slack notifications to avoid spamming users."""
 
@@ -73,21 +88,16 @@ def __init__(self, record_file: str) -> None:
         self._truncate_records()
 
     def _get_jwds(self) -> List[str]:
-        return [line[1] for line in self._read_records()]
-
-    def _read_records(self) -> List[List[str]]:
-        with open(self.record_file, "r") as f:
-            records = [
-                line.strip().split("\t") for line in f.readlines() if line.strip()
-            ]
-        return self._validate(records)
+        return [record.jwd for record in self._read_records()]
 
-    def _validate(self, records: List[List[str]]) -> List[List[str]]:
+    def _read_records(self) -> List[Record]:
         try:
-            for datestr, path in records:
-                if not isinstance(datestr, str) and isinstance(path, str):
-                    raise ValueError
-                datetime.fromisoformat(datestr)
+            with open(self.record_file, "r") as f:
+                records = [
+                    Record(*line.strip().split("\t"))
+                    for line in f.readlines()
+                    if line.strip()
+                ]
         except ValueError:
             logger.warning(
                 f"Invalid records found in {self.record_file}. The"
@@ -98,7 +108,7 @@ def _validate(self, records: List[List[str]]) -> List[List[str]]:
             return []
         return records
 
-    def _write_jwd(self, jwd: str) -> None:
+    def _write_record(self, jwd: str) -> None:
         with open(self.record_file, "a") as f:
             f.write(f"{datetime.now()}\t{jwd}\n")
 
@@ -110,17 +120,17 @@ def _truncate_records(self) -> None:
         """Truncate older records."""
         records = self._read_records()
         with open(self.record_file, "w") as f:
-            for datestr, jwd_path in records:
-                if datetime.fromisoformat(datestr) > datetime.now() - timedelta(
+            for record in records:
+                if datetime.fromisoformat(record.date) > datetime.now() - timedelta(
                     days=SLACK_NOTIFY_PERIOD_DAYS
                 ):
-                    f.write(f"{datestr}\t{jwd_path}\n")
+                    f.write(f"{record.date}\t{record.jwd}\n")
 
     def contains(self, jwd: Union[pathlib.Path, str]) -> bool:
         jwd = str(jwd)
         exists = jwd in self._get_jwds()
         if not exists:
-            self._write_jwd(jwd)
+            self._write_record(jwd)
         return exists