Refactor: separate CSV table into new module

Database is now subclass of CsvTable

1 files changed, 19 insertions, 81 deletions

diff --git a/jimbrella/database.py b/jimbrella/database.py
index e43b8f2..07ff6e0 100644
--- a/jimbrella/database.py
+++ b/jimbrella/database.py
@@ -1,15 +1,16 @@
 import csv
 import os
 from datetime import datetime, timedelta
+from .csv_table import CsvTable
 from .lockfile import Lockfile
-from .utils import human_datetime, human_timedelta
+from .utils import identity, human_datetime, human_timedelta
 from .config import DUE_HOURS
 from .exceptions import *
 
 STATUSES = ["available", "lent", "overdue", "maintenance", "withheld", "unknown"]
 
 
-class Database:
+class Database(CsvTable):
     def __init__(self, path):
         """A database of all umbrellas and their current state.
 
@@ -43,86 +44,23 @@ class Database:
         - lent_at       | (date) if status is "lent", lent_at is the submission time of the user's
                         | jForm answer sheet. is blank otherwise.
         """
-        self.path = path
-        self.lockfile = Lockfile(self.path)
-
-    def _read(self) -> list:
-        """Deserialize database."""
-        # Create database if it does not yet exist
-        try:
-            f = open(self.path, "x")
-            f.close()
-        except FileExistsError:
-            pass
-
-        with open(self.path) as f:
-            reader = csv.reader(f)
-            umbrellas = [
+        super().__init__(
+            path,
+            [
+                {"name": "serial", "deserializer": int},
+                {"name": "alias"},
+                {"name": "status"},
+                {"name": "tenant_name"},
+                {"name": "tenant_id"},
+                {"name": "tenant_phone"},
+                {"name": "tenant_email"},
                 {
-                    "serial": int(row[0]),
-                    "alias": row[1],
-                    "status": row[2],
-                    "tenant_name": row[3],
-                    "tenant_id": row[4],
-                    "tenant_phone": row[5],
-                    "tenant_email": row[6],
-                    "lent_at": datetime.fromisoformat(row[7]) if row[7] else None,
-                }
-                for row in reader
-            ]
-            f.close()
-
-            return umbrellas
-
-    def _write(self, umbrellas: list) -> None:
-        """Serialize database. When a failure occurs, abort and recover data."""
-        # make backup in memory
-        with open(self.path) as f:
-            backup = f.read()
-            f.close()
-
-        self.lockfile.lock()
-
-        f = open(self.path, "w")
-        try:
-            writer = csv.writer(f)
-            writer.writerows(
-                [
-                    [
-                        umb["serial"],
-                        umb["alias"],
-                        umb["status"],
-                        umb["tenant_name"],
-                        umb["tenant_id"],
-                        umb["tenant_phone"],
-                        umb["tenant_email"],
-                        umb["lent_at"].isoformat(timespec="seconds")
-                        if umb["lent_at"]
-                        else "",
-                    ]
-                    for umb in umbrellas
-                ]
-            )
-        except Exception as e:
-            # failure occurred on write
-            # abort write, and write back contents as they were before
-            # TODO: pass on exception and keep log
-            f.close()
-            f = open(self.path, "w")
-            f.write(backup)
-            f.close()
-            raise e
-
-        self.lockfile.unlock()
-
-    def _update(self, update: dict) -> list:
-        """Update status of one umbrella, and return the entire updated database."""
-        umbrellas = self._read()
-        for idx, umb in enumerate(umbrellas):
-            if umb["serial"] == update["serial"]:
-                umbrellas[idx] = update
-                self._write(umbrellas)
-                return umbrellas
+                    "name": "lent_at",
+                    "serializer": lambda d: d.isoformat(timespec="seconds") if d else "",
+                    "deserializer": lambda d: datetime.fromisoformat(d) if d else None,
+                },
+            ],
+        )
 
     def _find_by_serial(self, serial: int) -> dict:
         """Given a serial number, returns an umbrella with such serial.