import sqlite3 from datetime import datetime from typing import Union from .config import config class AdminLog: """Logs admin operations in a database so they can be recalled and possibly undone. Every time an admin modifies the Umbrellas table via the web console, the information below are kept for archival and/or rollback. Once created, the log cannot be changed. - serial | incremental serial number for each modification - date | modification time, same format as Umbrellas table - actor | admin username - id | umbrella id - (status|tenant_(name|id|phone|email)|lent_at)_a: respective values in database pertaining to umbrella # before modification - (status|tenant_(name|id|phone|email)|lent_at)_b: same as above but after modification - note | optional note describing the modification Schema: CREATE TABLE AdminLog( serial INT PRIMARY KEY, date TEXT, actor TEXT, id INT, status_a TEXT, status_b TEXT, tenant_name_a TEXT, tenant_name_b TEXT, tenant_id_a TEXT, tenant_id_b TEXT, tenant_phone_a TEXT, tenant_phone_b TEXT, tenant_email_a TEXT, tenant_email_b TEXT, lent_at_a TEXT, lent_at_b TEXT, note TEXT ); """ def __init__(self, path: str): self.path = path def read(self, maxlogs=20, serial=None) -> Union[dict, list]: db = sqlite3.connect(self.path) db.row_factory = sqlite3.Row if serial is None: data = db.execute("SELECT * FROM AdminLog ORDER BY serial DESC").fetchmany( maxlogs ) else: data = db.execute( "SELECT * FROM AdminLog WHERE serial = ?", (serial,) ).fetchone() db.close() return data def log( self, date: str, actor: str, id: int, umb_a: dict, umb_b: dict, note="" ) -> None: """Add an entry to log table. Does no authentication whatsoever. Authenticity must be ensured before calling this method. `date` must be an ISO 8601 string. """ db = sqlite3.connect(self.path) # get serial of most recent log, add one latest = db.execute( "SELECT serial FROM AdminLog ORDER BY serial DESC" ).fetchone() if latest is None: serial = 1 else: serial = latest[0] + 1 db.execute( "INSERT INTO AdminLog VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", ( serial, date, actor, id, umb_a.get("status", ""), umb_b.get("status", ""), umb_a.get("tenant_name", ""), umb_b.get("tenant_name", ""), umb_a.get("tenant_id", ""), umb_b.get("tenant_id", ""), umb_a.get("tenant_phone", ""), umb_b.get("tenant_phone", ""), umb_a.get("tenant_email", ""), umb_b.get("tenant_email", ""), umb_a.get("lent_at", ""), umb_b.get("lent_at", ""), note, ), ) db.commit() db.close()