nightly - 2025-09-20
This commit is contained in:
1
api/repos/__init__.py
Normal file
1
api/repos/__init__.py
Normal file
@@ -0,0 +1 @@
|
||||
from .api_keys import ApiKey, ApiKeyRepo
|
||||
43
api/repos/api_keys.py
Normal file
43
api/repos/api_keys.py
Normal file
@@ -0,0 +1,43 @@
|
||||
import random as r
|
||||
import string
|
||||
from dataclasses import dataclass
|
||||
from .template import Repo
|
||||
|
||||
rdm_set = string.ascii_lowercase + string.digits
|
||||
|
||||
|
||||
@dataclass
|
||||
class ApiKey:
|
||||
api_key: str
|
||||
pc_name: str = ""
|
||||
|
||||
|
||||
class ApiKeyRepo(Repo):
|
||||
def check_api(self, api_key: str) -> bool:
|
||||
self.cur.execute("SELECT * FROM api_keys WHERE api_key = %s", (api_key,))
|
||||
result = self.cur.fetchone()
|
||||
if result:
|
||||
return True
|
||||
else:
|
||||
return False
|
||||
|
||||
def create_api(self, name: str) -> str:
|
||||
while True:
|
||||
new_key = "".join(r.choice(rdm_set) for i in range(16))
|
||||
if not self.check_api(new_key):
|
||||
break
|
||||
self.cur.execute("INSERT INTO api_keys VALUES (%s, %s)", (new_key, name))
|
||||
self.conn.commit()
|
||||
return new_key
|
||||
|
||||
def get_all(self) -> list[ApiKey]:
|
||||
self.cur.execute("SELECT * FROM api_keys")
|
||||
results = self.cur.fetchall()
|
||||
if not results:
|
||||
return []
|
||||
return [ApiKey(*r) for r in results]
|
||||
|
||||
def delete(self, api_key: str) -> str:
|
||||
self.cur.execute("DELETE FROM api_keys WHERE api_key = %s", (api_key,))
|
||||
self.conn.commit()
|
||||
return "Key deleted successfully."
|
||||
39
api/repos/prefixes.py
Normal file
39
api/repos/prefixes.py
Normal file
@@ -0,0 +1,39 @@
|
||||
from .template import Repo
|
||||
from dataclasses import dataclass
|
||||
from exceptions import not_found
|
||||
|
||||
|
||||
@dataclass
|
||||
class Prefix:
|
||||
name: str
|
||||
color: str = ""
|
||||
weight: int = 0
|
||||
|
||||
|
||||
class PrefixRepo(Repo):
|
||||
def get_all(self) -> list[Prefix]:
|
||||
self.cur.execute("SELECT * FROM prefixes ORDER BY weight, name")
|
||||
results = self.cur.fetchall()
|
||||
if not results:
|
||||
return []
|
||||
return [Prefix(*r) for r in results]
|
||||
|
||||
def get_one(self, prefix_name: str):
|
||||
self.cur.execute("SELECT * FROM prefixes WHERE name = %s", (prefix_name,))
|
||||
result = self.cur.fetchone()
|
||||
if not result:
|
||||
raise not_found
|
||||
return Prefix(*result)
|
||||
|
||||
def add_one(self, prefix: Prefix) -> str:
|
||||
self.cur.execute(
|
||||
"REPLACE INTO prefixes VALUES (%s, %s, %s)",
|
||||
(prefix.name, prefix.color, prefix.weight),
|
||||
)
|
||||
self.conn.commit()
|
||||
return "Prefix inserted successfully."
|
||||
|
||||
def del_one(self, prefix_name: str) -> str:
|
||||
self.cur.execute("DELETE FROM prefixes WHERE name = %s", (prefix_name,))
|
||||
self.conn.commit()
|
||||
return "Prefix deleted successfully."
|
||||
6
api/repos/template.py
Normal file
6
api/repos/template.py
Normal file
@@ -0,0 +1,6 @@
|
||||
from db import session
|
||||
|
||||
|
||||
class Repo:
|
||||
def __init__(self):
|
||||
self.conn, self.cur = session()
|
||||
Reference in New Issue
Block a user