90 lines
2.2 KiB
Python
90 lines
2.2 KiB
Python
from contextlib import contextmanager
|
|
import duckdb
|
|
import bcrypt
|
|
from rich import print
|
|
|
|
|
|
from lib import settings
|
|
from models.user import UserIn, UserOut
|
|
|
|
|
|
@contextmanager
|
|
def get_db():
|
|
try:
|
|
conn = duckdb.connect(database=settings.db_url)
|
|
yield conn
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def db_run(sql):
|
|
with get_db() as conn:
|
|
conn.execute(sql)
|
|
|
|
|
|
def init_db():
|
|
if settings.db_url.exists():
|
|
# TODO: Development feature. Remove this in production
|
|
from os import unlink
|
|
|
|
unlink(settings.db_url)
|
|
# return
|
|
sql = ""
|
|
with open("lib/database.sql") as f:
|
|
sql = f.read()
|
|
|
|
db_run(sql)
|
|
|
|
# Initialize default user
|
|
|
|
hashed_password = bcrypt.hashpw("password".encode("utf-8"), bcrypt.gensalt())
|
|
|
|
sql = f"INSERT INTO users (username, password, is_admin) VALUES ('admin', '{hashed_password.decode('utf-8')}', true)"
|
|
db_run(sql)
|
|
|
|
sql = f"INSERT INTO users (username, password, is_admin) VALUES ('test', '{hashed_password.decode('utf-8')}', false)"
|
|
db_run(sql)
|
|
|
|
|
|
def get_user_by_username(username) -> UserIn | None:
|
|
sql = f"SELECT * FROM users WHERE username = '{username}'"
|
|
with get_db() as conn:
|
|
result = conn.execute(sql).fetchone()
|
|
if result is not None:
|
|
return UserIn(
|
|
id=str(result[0]),
|
|
username=result[1],
|
|
password=result[2],
|
|
email=result[3],
|
|
avatar_url=result[4],
|
|
is_active=result[5],
|
|
is_admin=result[6],
|
|
last_login=result[7],
|
|
created_at=result[8],
|
|
updated_at=result[9],
|
|
)
|
|
|
|
return None
|
|
|
|
|
|
def get_all_users() -> list[UserOut]:
|
|
sql = "SELECT * FROM users"
|
|
with get_db() as conn:
|
|
result = conn.execute(sql).fetchall()
|
|
users = []
|
|
for row in result:
|
|
user = UserOut(
|
|
id=str(row[0]),
|
|
username=row[1],
|
|
email=row[3],
|
|
avatar_url=row[4],
|
|
is_active=row[5],
|
|
is_admin=row[6],
|
|
last_login=row[7],
|
|
created_at=row[8],
|
|
updated_at=row[9],
|
|
)
|
|
users.append(user)
|
|
|
|
return users
|