Add mp.py

Signed-off-by: Ari Archer <ari@ari.lt>

.gitignore

@@ -168,3 +168,5 @@ cython_debug/
 
 *.env
 *.key
+
+/status

src/aw/mp.py

@@ -0,0 +1,55 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+"""Multiprocessing helpers"""
+
+
+import os
+import pickle
+import typing as t
+from collections import UserDict
+
+
+class FSDict(UserDict[t.Hashable, t.Any]):
+    """Filesystem-based dict."""
+
+    def __init__(self, filename: str, d: t.Optional[t.Dict[t.Hashable, t.Any]] = None):
+        super().__init__()
+
+        self.filename: str = filename
+        self.data: t.Dict[t.Hashable, t.Any] = d if d else {}
+
+        if not os.path.exists(self.filename):
+            self.save_data()
+
+        self.last_modified: float = self.get_last_modified()
+        self.load_data()
+
+    def get_last_modified(self) -> float:
+        """Return the last modified time of the JSON file."""
+
+        return os.path.getmtime(self.filename)
+
+    def load_data(self) -> None:
+        """Load data from the JSON file."""
+
+        with open(self.filename, "rb") as fp:
+            self.data = pickle.load(fp)
+
+    def save_data(self) -> None:
+        """Save current data to the JSON file."""
+
+        with open(self.filename, "wb") as fp:
+            pickle.dump(self.data, fp)
+
+    def __getitem__(self, key: t.Hashable) -> t.Any:  # type: ignore
+        current_modified: float = self.get_last_modified()
+
+        if current_modified != self.last_modified:
+            self.load_data()
+            self.last_modified = current_modified
+
+        return super().__getitem__(key)  # type: ignore
+
+    def __setitem__(self, key: t.Hashable, value: t.Any) -> None:
+        super().__setitem__(key, value)  # type: ignore
+        self.save_data()

src/aw/views.py

@@ -6,25 +6,24 @@ import datetime
 import hashlib
 import os
 import typing as t
-from multiprocessing import Manager
 
 import flask
 import validators
 from flask_limiter.util import get_remote_address
 from werkzeug.wrappers import Response
 
-from . import email, models, util
+from . import email, models, mp, util
 from .c import c
 from .limiter import limiter
 from .routing import Bp
 
 views: Bp = Bp("views", __name__)
-manager = Manager()
-status = manager.dict(
+status: mp.FSDict = mp.FSDict(
+    "status",
     {
         "status": "<i>No status</i>",
         "last_updated": datetime.datetime.now(datetime.timezone.utc),
-    }
+    },
 )