db.py (view raw)
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 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 |
import aiosqlite
from aiosqlite import Error
class Database:
def __init__(self, db_name: str):
self.db_name = db_name
# ====================== INITIALIZATION ======================== #
async def initial_setup(self):
await self.create_tables()
await self.create_id_rows()
# could be done more flexibly somehow probably, receiving the schema from the logic.py, but ehhhhhhhhhhh (also not databse independent then)
async def create_tables(self):
async with aiosqlite.connect(self.db_name) as db:
await db.execute("""CREATE TABLE IF NOT EXISTS clubs (
id INTEGER PRIMARY KEY,
channel_name TEXT NOT NULL UNIQUE,
owner_id INTEGER NOT NULL UNIQUE,
role_id INTEGER NOT NULL UNIQUE,
role_name TEXT NOT NULL UNIQUE);""")
await db.commit()
await db.execute("""CREATE TABLE IF NOT EXISTS members (
id INTEGER PRIMARY KEY,
user_id INTEGER NOT NULL,
club_id INTEGER NOT NULL,
UNIQUE(user_id, club_id) ON CONFLICT REPLACE,
FOREIGN KEY (club_id)
REFERENCES clubs (id)
ON UPDATE CASCADE
ON DELETE CASCADE);""")
await db.commit()
await db.execute("""CREATE TABLE IF NOT EXISTS existing_roles (
id INTEGER PRIMARY KEY,
roles TEXT NOT NULL UNIQUE);""")
await db.commit()
await db.execute("""CREATE TABLE IF NOT EXISTS ids (
id INTEGER PRIMARY KEY,
id_type TEXT UNIQUE NOT NULL,
discord_id INTEGER);""")
await db.commit()
async def create_id_rows(self):
id_types = [
"booster_role_id",
"distributor_channel_id",
"new_channel_category_id",
"club_role_header_role_id",
]
for id_type in id_types:
async with aiosqlite.connect(self.db_name) as db:
await db.execute(
"""INSERT OR IGNORE INTO ids (id_type) VALUES(?);""", (id_type,)
)
await db.commit()
# ========================= SETUP ============================ #
async def get_discord_id(self, id_type: str) -> Error | None:
async with aiosqlite.connect(self.db_name) as db:
async with db.execute(
"SELECT discord_id FROM ids WHERE id_type = ?;", (id_type,)
) as cursor:
async for row in cursor:
return row[0]
async def add_id(self, id_type: str, discord_id: int):
async with aiosqlite.connect(self.db_name) as db:
try:
await db.execute(
"""UPDATE ids
SET discord_id = ?
WHERE id_type = ?;""",
(discord_id, id_type),
)
await db.commit()
except Error as e:
print(e)
return e
return None
async def get_discord_ids(self):
async with aiosqlite.connect(self.db_name) as db:
async with db.execute("SELECT * FROM ids;") as cursor:
result = []
async for row in cursor:
result.append(row)
return result
# =========================== CREATE CLUB =========================== #
async def get_booster_role_id(self) -> int | None:
async with aiosqlite.connect(self.db_name) as db:
async with db.execute(
"SELECT discord_id FROM ids WHERE id_type = 'booster_role_id';"
) as cursor:
id = await cursor.fetchone()
if id:
return id[0]
else:
return None
async def select_role_id_by_owner(self, owner_id):
async with aiosqlite.connect(self.db_name) as db:
try:
async with db.execute(
"SELECT role_id FROM clubs WHERE owner_id = ?;", (owner_id,)
) as cursor:
return await cursor.fetchone()[3]
except Error as e:
print(e)
return e
async def select_club_by_channel_name(self, channel_name):
async with aiosqlite.connect(self.db_name) as db:
async with db.execute(
"SELECT * FROM clubs WHERE channel_name = ?;", (channel_name,)
) as cursor:
async for row in cursor:
return row
async def select_club_by_role_name(self, role_name):
async with aiosqlite.connect(self.db_name) as db:
async with db.execute(
"SELECT * FROM clubs WHERE role_name = ?;", (role_name,)
) as cursor:
async for row in cursor:
return row
async def create_club(
self, channel_name: str, owner: int, role_id: int, role_name: str
):
print(
f"DB: create_club received:\n channel_name: {channel_name}\n owner: {owner}\n role_id: {role_id}"
)
args = (channel_name, owner, role_id, role_name)
sql = """INSERT INTO clubs (channel_name,owner_id,role_id,role_name)
VALUES(?,?,?,?);"""
try:
async with aiosqlite.connect(self.db_name) as db:
await db.execute(sql, args)
await db.commit()
except Error as e:
return e
# ============================ EDIT CLUB ========================== #
async def club_edit(self, owner_id: int, column: str, value: str):
print(
f"DB: updating club:\n owner_id: {owner_id}\n column: {column}\n value: {value}"
)
async with aiosqlite.connect(self.db_name) as db:
try:
await db.execute(
f"UPDATE clubs SET {column} = ? WHERE owner_id = ?;",
(value, owner_id),
)
await db.commit()
return None
except Error as e:
print(e)
return e
# ============================ add member ================================== #
async def add_member(self, member: int, owner: int):
sql = """INSERT INTO members (user_id, club_id)
VALUES(?,?);"""
try:
async with aiosqlite.connect(self.db_name) as db:
async with db.execute(
"SELECT id FROM clubs WHERE owner_id = ?;", (owner,)
) as cursor:
async for row in cursor:
args = (member, row[0])
await db.execute(sql, args)
await db.commit()
except Error as e:
print(e)
return e
async def select_role_id_by_owner(self, member: int):
async with aiosqlite.connect(self.db_name) as db:
async with db.execute(
"SELECT role_id FROM clubs WHERE owner_id = ?;", (member,)
) as cursor:
async for row in cursor:
return row[0]
# ============================ remove member ================================== #
async def remove_member(self, member: int, owner: int):
club_id = await self.get_club_id_by_owner(owner)
try:
async with aiosqlite.connect(self.db_name) as db:
await db.execute(
"DELETE FROM members WHERE user_id = ? AND club_id = ?;",
(member, club_id),
)
await db.commit()
except Error as e:
print(e)
return e
# ============================= list member clubs ========================== #
async def select_clubs_of_member(self, member: int) -> list:
clubs = []
try:
async with aiosqlite.connect(self.db_name) as db:
async with db.execute(
"SELECT club_id FROM members WHERE user_id = ?;", (member,)
) as cursor:
async for row in cursor:
async with db.execute(
"SELECT channel_name, role_id FROM clubs WHERE id = ?;",
(row[0],),
) as cursor2:
async for row2 in cursor2:
clubs.append(row2)
except Error as e:
print(e)
return clubs
# ============================= list random stuff ================================ #
async def list_tables(self):
async with aiosqlite.connect(self.db_name) as db:
async with db.execute(
"SELECT name FROM sqlite_master WHERE type = 'table';"
) as cursor:
async for row in cursor:
print(row)
async def select_clubs(self) -> list:
clubs = []
async with aiosqlite.connect(self.db_name) as db:
async with db.execute("SELECT * FROM clubs;") as cursor:
async for row in cursor:
clubs.append(row)
return clubs
async def get_channel_name_role_name_by_member(self, user_id) -> list:
clubs = []
async with aiosqlite.connect(self.db_name) as db:
async with db.execute(
"SELECT club_id FROM members WHERE user_id = ?;", (user_id,)
) as cursor:
async for club_id in cursor:
async with db.execute(
"SELECT channel_name, role_id, id FROM clubs WHERE id = ?;",
(club_id[0],),
) as cursor2:
async for club in cursor2:
clubs.append(club)
return clubs
async def get_owner_by_club_id(self, club_id):
async with aiosqlite.connect(self.db_name) as db:
async with db.execute(
"SELECT owner_id FROM clubs WHERE id = ?;", (club_id,)
) as cursor:
result = await cursor.fetchone()
return result[0]
async def get_club_id_by_owner(self, owner: int):
async with aiosqlite.connect(self.db_name) as db:
async with db.execute(
"SELECT id FROM clubs WHERE owner_id = ?;", (owner,)
) as cursor:
result = await cursor.fetchone()
return result[0]
async def delete_club(self, owner_id):
async with aiosqlite.connect(self.db_name) as db:
await db.execute("DELETE FROM clubs WHERE owner_id = ?", (owner_id,))
club_id = await self.get_club_id_by_owner(owner_id)
await db.execute("DELETE FROM members WHERE club_id = ?", (club_id,))
await db.commit()
async def get_channel_name_by_owner(self, owner_id):
async with aiosqlite.connect(self.db_name) as db:
async with db.execute(
"SELECT channel_name FROM clubs WHERE owner_id = ?;", (owner_id,)
) as cursor:
result = await cursor.fetchone()
return result[0]
async def get_header_role(self):
async with aiosqlite.connect(self.db_name) as db:
async with db.execute(
"SELECT discord_id FROM ids WHERE id_type LIKE 'club_role_header_role_id';"
) as cursor:
result = await cursor.fetchone()
return result[0]
|