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
|
import sqlite3
from logging import Logger, getLogger
from sqlite3 import PARSE_DECLTYPES, Connection, Cursor
from typing import Any, Sequence
from pyssg.db.tuple import adapt_tuple, convert_tuple
from pyssg.db.queries import CREATE_FILES_TABLE, SELECT_FILE, SELECT_FILE_ALL, INSERT_FILE, UPDATE_FILE, UPDATE_FILE_TAGS
log: Logger = getLogger(__name__)
class Database:
def __init__(self, path: str) -> None:
sqlite3.register_adapter(tuple, adapt_tuple)
sqlite3.register_converter("tuple", convert_tuple)
self.con: Connection = sqlite3.connect(path, detect_types=PARSE_DECLTYPES)
self.cur: Cursor = self.con.cursor()
# create statements are always commited
self.query(CREATE_FILES_TABLE)
# commits the transactions, closes connection and cursor
def write(self) -> None:
self.con.commit()
self.cur.close()
self.con.close()
def query(self, sql: str,
params: dict | Sequence = ()) -> list[Any]:
return self.cur.execute(sql, params).fetchall()
# commit query, doesn't wait until calling con.commit()
def cquery(self, sql: str,
params: dict | Sequence = ()) -> list[Any]:
out: list[Any]
with self.con:
out = self.query(sql, params)
return out
def select(self, fname: str) -> tuple | None:
out: list[Any]
out = self.query(SELECT_FILE, (fname,))
log.debug("select %s", out)
return out[0] if out else None
def select_all(self) -> list[Any] | None:
out: list[Any] = self.query(SELECT_FILE_ALL)
log.debug("select_all %s", out)
return out if out else None
def insert(self, fname: str,
ctime: float,
checksum: str,
tags: tuple | None = None) -> tuple:
params: tuple = (fname, ctime, checksum, tags)
out: tuple = self.query(INSERT_FILE, params)[0]
log.debug("insert %s", out)
return out
def update(self, fname: str,
mtime: float,
checksum: str,
tags: tuple | None = None) -> tuple:
params: tuple = (mtime, checksum, tags, fname)
out: tuple = self.query(UPDATE_FILE, params)[0]
log.debug("update %s", out)
return out
def update_tags(self, fname: str,
tags: tuple | None = None) -> tuple:
params: tuple = (tags, fname)
out: tuple = self.query(UPDATE_FILE_TAGS, params)[0]
log.debug("update_tags %s", out)
return out
|
