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
|
from dataclasses import dataclass
import pymysql
import random
import app
@dataclass
class Database:
safeLogin:bool = True #automatically login with the user in the config file, who is read only
user:str = None #otherwise, login with the given username and passwd
passwd:str = None
def __enter__(self):
if self.safeLogin:
self.__connection = pymysql.connect(
**app.CONFIG["mysql"],
charset = "utf8mb4"
)
else:
self.__connection = pymysql.connect(
user = self.user,
passwd = self.passwd,
host = app.CONFIG["mysql"]["host"],
db = app.CONFIG["mysql"]["db"],
charset = "utf8mb4"
)
return self
def __exit__(self, type, value, traceback):
self.__connection.close()
def get_header_links(self):
with self.__connection.cursor() as cursor:
cursor.execute("SELECT name, link FROM headerLinks ORDER BY name;")
return cursor.fetchall()
def get_image(self, imageName):
with self.__connection.cursor() as cursor:
cursor.execute("SELECT alt, url FROM images WHERE imageName = %s;", (imageName, ))
return cursor.fetchone()
def get_pfp_image(self):
with self.__connection.cursor() as cursor:
cursor.execute("SELECT alt, url FROM images WHERE pfp_img = 1;")
return random.choice(cursor.fetchall())
def get_header_articles(self):
with self.__connection.cursor() as cursor:
cursor.execute("SELECT articleName, link FROM headerArticles;")
return cursor.fetchall()
def get_all_categories(self):
with self.__connection.cursor() as cursor:
cursor.execute("SELECT category_name FROM categories;")
return [i[0] for i in cursor.fetchall()]
def add_category(self, category):
if not category in self.get_all_categories():
with self.__connection.cursor() as cursor:
cursor.execute("INSERT INTO categories (category_name) VALUES (%s);", (category, ))
self.__connection.commit()
return True
return False
def add_thought(self, category, title, markdown):
with self.__connection.cursor() as cursor:
cursor.execute("""
INSERT INTO thoughts (category_id, title, markdown_text)
VALUES ((
SELECT category_id FROM categories WHERE category_name = %s
), %s, %s);""", (category, title, markdown))
self.__connection.commit()
def get_thought(self, id_):
with self.__connection.cursor() as cursor:
cursor.execute("""
SELECT categories.category_name, thoughts.title, thoughts.dt, thoughts.markdown_text
FROM thoughts INNER JOIN categories
ON thoughts.category_id = categories.category_id
WHERE thought_id = %s;""", (id_, ))
return cursor.fetchone()
def get_categories_not(self, category_name):
with self.__connection.cursor() as cursor:
cursor.execute("SELECT category_name FROM categories WHERE category_name != %s;", (category_name, ))
return [i[0] for i in cursor.fetchall()]
def get_all_thoughts(self):
with self.__connection.cursor() as cursor:
cursor.execute("""
SELECT thought_id, title, dt, category_name FROM thoughts
INNER JOIN categories ON thoughts.category_id = categories.category_id;
""")
return cursor.fetchall()
if __name__ == "__main__":
with Database() as db:
print(db.get_all_thoughts())
|
