diff options
| -rw-r--r-- | jimbrella/admin.py | 19 | ||||
| -rw-r--r-- | jimbrella/admin_log.py | 1 | ||||
| -rw-r--r-- | jimbrella/umbrellas.py | 49 |
3 files changed, 36 insertions, 33 deletions
diff --git a/jimbrella/admin.py b/jimbrella/admin.py index b150e3d..9d7d98b 100644 --- a/jimbrella/admin.py +++ b/jimbrella/admin.py @@ -12,7 +12,7 @@ from .utils import human_datetime, human_timedelta, CST bp = Blueprint("admin", __name__, url_prefix="/admin") db = Umbrellas(config.get("general", "db_path")) users = Users(config.get("general", "db_path")) -admin_log = AdminLog(config.get("logging", "admin_log_path")) +admin_log = AdminLog(config.get("general", "db_path")) @bp.before_request @@ -115,7 +115,7 @@ def umbrellas_edit(): data[key] = request.form.get(key) try: - diff = db.update(data) + db.admin_modify(session["username"], data) except UmbrellaValueError as e: # invalid field is in `e.message`. return redirect( @@ -126,23 +126,10 @@ def umbrellas_edit(): except UmbrellaNotFoundError: abort(400) - for column, value_pair in diff.items(): - past, new = value_pair - admin_log.log( - "ADMIN_MODIFY_DB", - { - "admin_name": session["username"], - "id": data["id"], - "column": column, - "past_value": past, - "new_value": new, - }, - ) - return redirect(url_for("admin.umbrellas")) @bp.route("/logs") def logs(): logs = admin_log.read() - return render_template("admin/logs.html", logs=logs) + return render_template("admin/admin_logs.html", logs=logs) diff --git a/jimbrella/admin_log.py b/jimbrella/admin_log.py index f576f42..36127b3 100644 --- a/jimbrella/admin_log.py +++ b/jimbrella/admin_log.py @@ -104,4 +104,5 @@ class AdminLog: note, ), ) + db.commit() db.close() diff --git a/jimbrella/umbrellas.py b/jimbrella/umbrellas.py index b3a261d..9b5c7b8 100644 --- a/jimbrella/umbrellas.py +++ b/jimbrella/umbrellas.py @@ -2,6 +2,7 @@ import sqlite3 from datetime import datetime, timezone, timedelta from dateutil.parser import isoparse from typing import Union +from .admin_log import AdminLog from .utils import human_datetime, human_timedelta, CST from .exceptions import * @@ -13,7 +14,8 @@ class Umbrellas: """A database of all umbrellas and their current state. Currently, the data are represented in a SQLite database. Only admins have access to the - database, and have the power to modify it arbitrarily. + database, and have the power to modify it arbitrarily. Each time a modification is made, + AdminLog keeps a log. An SQL row for an umbrella consists of these columns: - id | int. unique identifier for the umbrella. @@ -42,11 +44,12 @@ class Umbrellas: ); """ self.path = path + self.admin_log = AdminLog(path) def read(self, umbid=None) -> Union[dict, list]: """Read umbrella data from database. - If umbid is an integer, returns dict pertaining to umbrella #<umbid>. + If umbid is an integer, returns dict pertaining to umbrella #<umbid>, or None if not found. If umbid is None, returns list of dicts for all umbrellas. """ db = sqlite3.connect(self.path) @@ -61,7 +64,7 @@ class Umbrellas: db.close() return data - def update(self, umb) -> dict: + def update(self, umb) -> None: """Update Umbrella table with new data given in `umb`. Not all fields in an umbrella dict need to be present in `umb`. Only `id` is required. @@ -95,15 +98,10 @@ class Umbrellas: if umb_in_db is None: raise UmbrellaNotFoundError(umbid) - diff = {} - status = umb_in_db["status"] if "status" in umb: if umb["status"] in STATUSES: status = umb["status"] - if umb_in_db["status"] != umb["status"]: - diff["status"] = (umb_in_db["status"], umb["status"]) - db.execute( "UPDATE Umbrellas SET status = ? WHERE id = ?", (status, umbid) ) @@ -118,9 +116,6 @@ class Umbrellas: "tenant_email", ): if col in umb: - if umb_in_db[col] != umb[col]: - diff[col] = (umb_in_db[col], umb[col]) - db.execute( f"UPDATE Umbrellas SET {col} = ? WHERE id = ?", ( @@ -143,11 +138,6 @@ class Umbrellas: if lent_at.tzinfo is None: lent_at = lent_at.replace(tzinfo=CST) - diff["lent_at"] = ( - umb_in_db["lent_at"], - lent_at.isoformat(timespec="milliseconds"), - ) - db.execute( "UPDATE Umbrellas SET lent_at = ? WHERE id = ?", ( @@ -169,7 +159,32 @@ class Umbrellas: # now that new data are validated, commit the SQL transaction db.commit() db.close() - return diff + + def admin_modify(self, actor: str, umb: dict, note="") -> None: + """Update db with `umb`, and also keep a log with AdminLog. + + Similar to AdminLog.log, this method does no authentication either. + """ + try: + umbid = umb["id"] + except KeyError: + raise UmbrellaValueError("id") + + umb_old = dict(self.read(umbid)) + if umb_old is None: + raise UmbrellaNotFoundError(umbid) + + self.update(umb) + umb_new = dict(self.read(umbid)) + if umb_old != umb_new: + self.admin_log.log( + datetime.now(tz=CST).isoformat(timespec="milliseconds"), + actor, + umbid, + umb_old, + umb_new, + note, + ) def take_away( self, umbid, date, tenant_name, tenant_id, tenant_phone="", tenant_email="" |