-
Notifications
You must be signed in to change notification settings - Fork 0
/
store.py
67 lines (52 loc) · 1.85 KB
/
store.py
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
import pathlib
import sqlite3
import json
from fetcher import HouseStore, House
_schema = """
CREATE TABLE IF NOT EXISTS houses(
id INTEGER PRIMARY KEY,
raw TEXT NOT NULL
);
"""
class LiteHouseStore(HouseStore):
__conn: sqlite3.Connection
def __init__(self, path: str):
self.__conn = sqlite3.connect(path)
self.__ensure_table()
def __ensure_table(self) -> None:
with self.__conn as conn:
conn.execute(_schema)
def __setitem__(self, key: int, value: House):
with self.__conn as conn:
conn.execute(
"INSERT OR REPLACE INTO houses (id, raw) VALUES (?, ?)",
(key, json.dumps(House.to_json(value)),),
)
def __getitem__(self, key: int):
with self.__conn as conn:
cursor = conn.execute("SELECT raw FROM houses WHERE id = ?", (key,))
results = cursor.fetchall()
if len(results) == 0:
raise KeyError("house not exists")
if len(results) != 1:
raise RuntimeError("database corrupted")
raw_data = results[0][0]
return House.from_json(json.loads(raw_data))
def __delitem__(self, key: int):
with self.__conn as conn:
cursor = conn.execute("DELETE FROM houses WHERE id = ?", (key,))
if cursor.rowcount != 1:
raise KeyError("house not exists")
def __len__(self):
with self.__conn as conn:
cursor = conn.execute("SELECT COUNT(id) FROM houses")
result = cursor.fetchone()
return result[0]
def __iter__(self):
with self.__conn as conn:
cursor = conn.execute("SELECT id from houses")
while True:
row = cursor.fetchone()
if row is None:
return
yield row[0]