import sqlite3 from datetime import datetime from typing import Union from .config import config class Logger: def __init__(self, path: str): """Read/write interface to non-volatile logger for admins to inspect. This module does no authentication whatsoever. Authenticity must be ensured beforehand. """ self.path = path def read(self, table, maxlogs=20, serial=None) -> Union[dict, list]: db = sqlite3.connect(self.path) db.row_factory = sqlite3.Row if serial is None: data = db.execute(f"SELECT * FROM {table} ORDER BY serial DESC").fetchmany( maxlogs ) else: data = db.execute( f"SELECT * FROM {table} WHERE serial = ?", (serial,) ).fetchone() db.close() return data def read_admin(self, maxlogs=20, serial=None) -> Union[dict, list]: return self.read("AdminLog", maxlogs, serial) def read_tenant(self, maxlogs=20, serial=None) -> Union[dict, list]: return self.read("TenantLog", maxlogs, serial) def log_admin( self, date: str, actor: str, umbid: int, umb_a: dict, umb_b: dict, note="" ) -> None: """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 ); """ 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, umbid, 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() def log_tenant(self, date: str, action: str, umbid: int, umb: dict) -> None: """Keeps a log each time an umbrella is borrowed, returned, or marked overdue. - serial, date | same as AdminLog - action | one of "borrow", "return", and "overdue" - id, tenant_* | same as respective columns in table Umbrellas Schema: CREATE TABLE TenantLog( serial INT PRIMARY KEY, date TEXT, action TEXT, id INT, tenant_name TEXT, tenant_id TEXT, tenant_phone TEXT, tenant_email TEXT ); """ db = sqlite3.connect(self.path) latest = db.execute( "SELECT serial FROM TenantLog ORDER BY serial DESC" ).fetchone() if latest is None: serial = 1 else: serial = latest[0] + 1 db.execute( "INSERT INTO TenantLog VALUES (?, ?, ?, ?, ?, ?, ?, ?)", ( serial, date, action, umbid, umb.get("tenant_name", ""), umb.get("tenant_id", ""), umb.get("tenant_phone", ""), umb.get("tenant_email", ""), ), ) db.commit() db.close()