80 lines
2.1 KiB
Python
80 lines
2.1 KiB
Python
|
from typing import Optional
|
||
|
|
|
||
|
|
from app.extensions import db_connection
|
||
|
|
|
||
|
|
from app.models.user_model import User
|
||
|
|
|
||
|
|
|
||
|
|
def fetch_by_username(username: str) -> Optional[User]:
|
||
|
|
cursor = db_connection.cursor(dictionary=True)
|
||
|
|
|
||
|
|
cursor.execute("select * from user where username = %s", (username,))
|
||
|
|
|
||
|
|
result = cursor.fetchone()
|
||
|
|
|
||
|
|
result_user = (
|
||
|
|
User(
|
||
|
|
user_id=result["id"],
|
||
|
|
username=result["username"],
|
||
|
|
displayname=result["displayname"],
|
||
|
|
email=result["email"],
|
||
|
|
password=result["password"],
|
||
|
|
role_id=result["role_id"],
|
||
|
|
creation_date=result["creation_date"],
|
||
|
|
)
|
||
|
|
if result
|
||
|
|
else None
|
||
|
|
)
|
||
|
|
|
||
|
|
return result_user
|
||
|
|
|
||
|
|
|
||
|
|
def fetch_by_id(user_id: int) -> Optional[User]:
|
||
|
|
cursor = db_connection.cursor(dictionary=True)
|
||
|
|
|
||
|
|
cursor.execute("select * from user where id = %s", (user_id,))
|
||
|
|
|
||
|
|
result = cursor.fetchone()
|
||
|
|
result_user = (
|
||
|
|
User(
|
||
|
|
user_id=result["id"],
|
||
|
|
username=result["username"],
|
||
|
|
displayname=result["displayname"],
|
||
|
|
email=result["email"],
|
||
|
|
password=result["password"],
|
||
|
|
role_id=result["role_id"],
|
||
|
|
creation_date=result["creation_date"],
|
||
|
|
)
|
||
|
|
if result
|
||
|
|
else None
|
||
|
|
)
|
||
|
|
|
||
|
|
return result_user
|
||
|
|
|
||
|
|
|
||
|
|
def insert_user(new_user: User):
|
||
|
|
cursor = db_connection.cursor(dictionary=True)
|
||
|
|
|
||
|
|
cursor.execute(
|
||
|
|
"insert into user (username, displayname, email, password) values (%s, %s, %s, %s)",
|
||
|
|
(new_user.username, new_user.displayname, new_user.email, new_user.password),
|
||
|
|
)
|
||
|
|
db_connection.commit()
|
||
|
|
|
||
|
|
|
||
|
|
def delete_user(user: User):
|
||
|
|
cursor = db_connection.cursor(dictionary=True)
|
||
|
|
|
||
|
|
cursor.execute("delete from user where id = %s", (user.user_id,))
|
||
|
|
db_connection.commit()
|
||
|
|
|
||
|
|
|
||
|
|
def update_user(user: User):
|
||
|
|
cursor = db_connection.cursor(dictionary=True)
|
||
|
|
|
||
|
|
cursor.execute(
|
||
|
|
"update user set username=%s, displayname=%s, email=%s, password=%s where id = %s",
|
||
|
|
(user.username, user.displayname, user.email, user.password, user.user_id),
|
||
|
|
)
|
||
|
|
db_connection.commit()
|