summaryrefslogtreecommitdiff
path: root/jimbrella/database.py
diff options
context:
space:
mode:
authorFrederick Yin <fkfd@fkfd.me>2022-02-01 12:43:25 +0800
committerFrederick Yin <fkfd@fkfd.me>2022-02-01 12:43:25 +0800
commit0e47cf9a8d06e896c5197cb28cb5a2a518d255d1 (patch)
tree22a0267dbfe204038aaac48d8596d436b5d56291 /jimbrella/database.py
parent444966b2ff4a04374856d3a6759bef9e9f451c03 (diff)
SQLite in favor of CSV database
Deprecate csv database format, move around some methods
Diffstat (limited to 'jimbrella/database.py')
-rw-r--r--jimbrella/database.py252
1 files changed, 0 insertions, 252 deletions
diff --git a/jimbrella/database.py b/jimbrella/database.py
deleted file mode 100644
index 9b6e51f..0000000
--- a/jimbrella/database.py
+++ /dev/null
@@ -1,252 +0,0 @@
-from datetime import datetime, timedelta
-from .csv_table import CsvTable
-from .utils import human_datetime, human_timedelta
-from .config import DUE_HOURS, ADMIN_LOG_PATH
-from .exceptions import *
-
-STATUSES = ["available", "lent", "overdue", "maintenance", "withheld", "unknown"]
-
-
-class Database(CsvTable):
- def __init__(self, path):
- """A database of all umbrellas and their current state.
-
- Currently, the data are represented in a csv file.
-
- Explanation of data types:
- - id | a unique numerical identity of unsigned integer type.
- - uint | whatever fits the regex /[0-9]+/. despite carrying a numerical value, it is
- | internally represented as a string.
- - string | any UTF-8 string.
- - date | an ISO 8601 date of format "YYYY-MM-DDThh:mm:ss+08:00".
-
- The status of an umbrella consists of:
- - serial | (id) unique identifier for the umbrella.
- - alias | (string) future compatibility. a human readable (preferably cute) name
- | for a particular umbrella
- - status | (string) one of ("available", "lent", "overdue", "maintenance",
- | "withheld", "unknown")
- | available : is in service on the stand
- | lent : is in temporary possession of a user
- | overdue : is in overly prolonged possession of a user
- | maintenance : is on the stand but not in service
- | withheld : is not on the stand but rather in the possession of an
- | administrator
- | unknown : none of the above
- | NOTE: the values above are subject to changes.
- - tenant_name | (string) the person in temporary possession of the umbrella.
- - tenant_id | (uint) student or faculty ID.
- - tenant_phone | (uint) phone number via which to contact tenant when the lease is due.
- - tenant_email | (string) future compatibility. empty for the time being.
- - lent_at | (date) if status is "lent", lent_at is the submission time of the user's
- | jForm answer sheet. is blank otherwise.
- """
- super().__init__(
- path,
- [
- {"name": "serial", "deserializer": int},
- {"name": "alias"},
- {"name": "status"},
- {"name": "tenant_name"},
- {"name": "tenant_id"},
- {"name": "tenant_phone"},
- {"name": "tenant_email"},
- {
- "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.
- If there is no such umbrella, returns None.
- """
- umbrellas = self._read()
- for umb in umbrellas:
- if umb["serial"] == serial:
- return umb
-
- def read(self) -> list:
- """An interface to `_read()` with supplemental data included.
-
- All exposed methods with a return value should use this method instead of `_read()`.
-
- Supplemental data:
- - lent_at_str: string representation for lent_at.
- - lent_time_ago: time since umbrella was taken away by tenant. if umbrella is not
- taken away, its value is None.
- - lent_time_ago_str: string representation for lent_time_ago.
- """
- umbrellas = self._read()
- now = datetime.now()
- for idx, umb in enumerate(umbrellas):
- if umb["status"] in ("lent", "overdue"):
- umbrellas[idx]["lent_at_str"] = human_datetime(umb["lent_at"])
- lent_time_ago = now - umb["lent_at"]
- umbrellas[idx]["lent_time_ago"] = lent_time_ago
- umbrellas[idx]["lent_time_ago_str"] = human_timedelta(lent_time_ago)
- return umbrellas
-
- def update(self, umb) -> dict:
- """An interface to `_update()` with added convenience and safeguards.
-
- Convenience: Not all fields in an umbrella dict need to be present in `umb`. If an
- optional field is not there, its value is left untouched. Every field other
- than `serial` is optional.
-
- `serial` may be an int or a str that evaluates to the correct serial when
- int() is applied to it.
-
- `tenant_id` and `tenant_phone` may be a str or an int.
-
- `lent_at` in `umb` may either be a datetime.datetime object, or an ISO 8601
- formatted string.
-
- Safeguards: Invalid values are rejected as an Exception, in the default case, an
- UmbrellaValueError.
-
- Both: When a value is set, values of unneeded fields are automatically discarded
- if applicable. For example, when `status` is set to "available", `tenant_*`
- and `lent_at` are no longer needed; in fact, their existence is dangerous
- as it implies the status of the umbrella is "lent" or "overdue", which is
- not the case. Therefore, if `status` is set to "available", the unnecessary
- fields mentioned above are made blank, and these fields supplied in `umb`
- are ignored.
-
- Returns a dict, where the keys are all modified columns of the updated row, and the value
- of each is a 2-tuple (<past value>, <new value>).
- """
- # `serial` must be specified.
- if "serial" not in umb:
- raise UmbrellaValueError("serial")
-
- # check if umbrella #<serial> exists in database
- umb["serial"] = int(umb["serial"])
- umb_in_db = self._find_by_serial(umb["serial"])
- if umb_in_db is None:
- raise UmbrellaNotFoundError(umb["serial"])
-
- status = umb_in_db["status"]
- if "status" in umb and umb["status"]:
- if status not in STATUSES:
- raise UmbrellaValueError("status") # invalid
-
- # admin specifies a (perhaps different) status
- status = umb["status"]
-
- if "lent_at" in umb and umb["lent_at"]:
- # check if `umb` has valid date (`lent_at`) if any
- if isinstance(umb["lent_at"], datetime):
- lent_at = umb["lent_at"]
- elif isinstance(umb["lent_at"], str):
- try:
- umb["lent_at"] = datetime.fromisoformat(umb["lent_at"])
- except ValueError:
- raise UmbrellaValueError("lent_at")
- else:
- raise UmbrellaValueError("lent_at")
-
- # we will now check the validity of `umb` based on `status`
- if status == "available":
- # discard unneeded fields
- for key in ["tenant_name", "tenant_id", "tenant_phone", "tenant_email"]:
- umb[key] = ""
- umb["lent_at"] = None
- elif status in ("lent", "overdue"):
- # copy values in database into unfilled or non-existent fields of `umb`
- for key in [
- "tenant_name",
- "tenant_id",
- "tenant_phone",
- "tenant_email",
- "lent_at",
- ]:
- umb[key] = umb[key] if (key in umb and umb[key]) else umb_in_db[key]
-
- if not umb["lent_at"]:
- raise UmbrellaValueError("lent_at")
-
- # commit update
- self._update(umb)
-
- # find updated columns
- # essentially "diff umb_in_db umb"
- diff = {}
- for key, new in umb.items():
- past = umb_in_db[key]
- if any([new, past]) and new != past:
- # new and past are not both null values, and they are unequal
- diff[key] = (past, new)
-
- return diff
-
- @staticmethod
- def group_by_status(umbrellas) -> dict:
- """(static method) Returns umbrellas grouped into a dict by their status."""
- keys = set([umb["status"] for umb in umbrellas])
- # initiate statuses: each status is []
- statuses = dict.fromkeys(STATUSES, [])
- for key in keys:
- statuses[key] = [umb for umb in umbrellas if umb["status"] == key]
- return statuses
-
- @staticmethod
- def find_overdue(umbrellas) -> list:
- """(static method) Returns umbrellas in possession of their tenant for too long."""
- now = datetime.now()
- return [
- umb
- for umb in umbrellas
- if umb["lent_at"] is not None
- and now - umb["lent_at"] > timedelta(hours=DUE_HOURS)
- ]
-
- def take_away(
- self, serial, date, tenant_name, tenant_id, tenant_phone="", tenant_email=""
- ) -> None:
- """When a user has borrowed an umbrella."""
- umb = self._find_by_serial(serial)
- if umb is None:
- raise UmbrellaNotFoundError(serial)
- if umb["status"] != "available":
- raise UmbrellaStatusError
- umb["status"] = "lent"
- umb["tenant_name"] = tenant_name
- umb["tenant_id"] = tenant_id
- umb["tenant_phone"] = tenant_phone
- umb["tenant_email"] = tenant_email
- umb["lent_at"] = date
- self._update(umb)
-
- def give_back(self, serial, tenant_name, tenant_id) -> None:
- """When a user has returned an umbrella.
-
- `tenant_name` and `tenant_id` are used to verify if the umbrella is returned by the same
- person who borrowed it.
- """
- umb = self._find_by_serial(serial)
- if umb is None:
- raise UmbrellaNotFoundError(serial)
- if umb["status"] not in ("lent", "overdue"):
- raise UmbrellaStatusError
- if umb["tenant_name"] != tenant_name or umb["tenant_id"] != tenant_id:
- raise TenantIdentityError(umb["tenant_name"], tenant_name)
- umb["status"] = "available"
- for key in ["tenant_name", "tenant_id", "tenant_phone", "tenant_email"]:
- umb[key] = ""
- umb["lent_at"] = None
- self._update(umb)
-
- def mark_overdue(self, serial) -> None:
- """When an umbrella is overdue, change its status to "overdue"."""
- umb = self._find_by_serial(serial)
- if umb is None:
- raise UmbrellaNotFoundError(serial)
- elif umb["status"] != "lent":
- raise UmbrellaStatusError
- umb["status"] = "overdue"
- self._update(umb)