summaryrefslogtreecommitdiff
path: root/jimbrella/database.py
blob: e84165032269356d3bd8d4fcb93dd32f12712f71 (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
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
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)