diff options
Diffstat (limited to 'jimbrella/database.py')
-rw-r--r-- | jimbrella/database.py | 252 |
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) |