summaryrefslogtreecommitdiff
path: root/jimbrella/admin_log.py
blob: 36127b30d16ce25872c2af42151709bdba2edcb5 (plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
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 #<id> 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()