diff options
| -rw-r--r-- | jimbrella/__init__.py | 0 | ||||
| -rw-r--r-- | jimbrella/__main__.py | 10 | ||||
| -rw-r--r-- | jimbrella/database.py | 147 |
3 files changed, 157 insertions, 0 deletions
diff --git a/jimbrella/__init__.py b/jimbrella/__init__.py new file mode 100644 index 0000000..e69de29 --- /dev/null +++ b/jimbrella/__init__.py diff --git a/jimbrella/__main__.py b/jimbrella/__main__.py new file mode 100644 index 0000000..1e658c7 --- /dev/null +++ b/jimbrella/__main__.py @@ -0,0 +1,10 @@ +from .database import Database + +def test(): + db = Database("/home/fkfd/p/jimbrella/jimbrella.db.csv") + db.take_away(2, "Jim Brella", "1337", "10086") + db.give_back(1) + +if __name__ == "__main__": + test() + diff --git a/jimbrella/database.py b/jimbrella/database.py new file mode 100644 index 0000000..e841650 --- /dev/null +++ b/jimbrella/database.py @@ -0,0 +1,147 @@ +import csv +from datetime import datetime + + +class Database: + 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". default timezone is UTC+8. + + 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", "withheld", "maintenance", + | "withheld", "unknown") + | available : is in service on the stand + | lent : is in temporary 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. + """ + self.path = 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 = [ + { + "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.""" + with open(self.path) as f: + backup = f.read() + f.close() + + 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: + # 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() + + 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 + + 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 take_away( + self, serial, 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 ValueError(f"No umbrella with serial {serial} found.") + elif umb["status"] != "available": + raise ValueError(f"Umbrella with serial {serial} is inavailable.") + 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"] = datetime.now() + self._update(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 ValueError(f"No umbrella with serial {serial} found.") + elif umb["status"] != "lent": + raise ValueError(f"Umbrella with serial {serial} is not lent out.") + umb["status"] = "available" + for key in ["tenant_name", "tenant_id", "tenant_phone", "tenant_email"]: + umb[key] = "" + umb["lent_at"] = None + self._update(umb) |