from datetime import datetime, timedelta from .csv_table import CsvTable from .admin_log import AdminLog 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, }, ], ) self.admin_log = AdminLog(ADMIN_LOG_PATH) 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) -> list: """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 updated database (same as `_update()`). """ # `serial` must be specified. if "serial" not in umb: raise UmbrellaValueError("serial") # check if umbrella # 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") return self._update(umb) @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) elif 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) self.admin_log.log("TAKEAWAY", umb) def give_back(self, serial) -> None: """When a user has returned an umbrella.""" umb = self._find_by_serial(serial) if umb is None: raise UmbrellaNotFoundError(serial) elif umb["status"] not in ("lent", "overdue"): raise UmbrellaStatusError self.admin_log.log("GIVEBACK", umb) 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) self.admin_log.log("OVERDUE", umb)