204 lines
4.2 KiB
Python
204 lines
4.2 KiB
Python
import psycopg2.extras
|
|
from psycopg2._psycopg import connection
|
|
|
|
# user create and delete
|
|
|
|
def create_user(
|
|
conn: connection,
|
|
username: str,
|
|
password: str,
|
|
role: str = "user"
|
|
):
|
|
with conn.cursor() as cur:
|
|
cur.execute(
|
|
"""
|
|
insert into picrinth.users
|
|
(username, password, role, disabled, created_at)
|
|
values (%s, %s, %s, false, now())
|
|
""",
|
|
(username, password, role),
|
|
)
|
|
conn.commit()
|
|
return cur.rowcount > 0
|
|
|
|
|
|
def delete_user(
|
|
conn: connection,
|
|
username: str
|
|
):
|
|
with conn.cursor() as cur:
|
|
cur.execute(
|
|
"""
|
|
delete from picrinth.users
|
|
where username = %s
|
|
""",
|
|
(username,),
|
|
)
|
|
conn.commit()
|
|
return cur.rowcount > 0
|
|
|
|
|
|
# user checks
|
|
|
|
def check_user_existence(
|
|
conn: connection,
|
|
username: str
|
|
):
|
|
with conn.cursor() as cur:
|
|
cur.execute(
|
|
"""
|
|
select exists(
|
|
select 1
|
|
from picrinth.users
|
|
where username = %s
|
|
);
|
|
""",
|
|
(username,),
|
|
)
|
|
return cur.fetchone()[0] # type: ignore
|
|
|
|
def check_user_disabled(
|
|
conn: connection,
|
|
username: str
|
|
):
|
|
with conn.cursor() as cur:
|
|
cur.execute(
|
|
"""
|
|
select disabled
|
|
from picrinth.users
|
|
where username = %s;
|
|
""",
|
|
(username,),
|
|
)
|
|
result = cur.fetchone()
|
|
if result is None:
|
|
return None
|
|
return result[0] # type: ignore
|
|
|
|
|
|
# user updates
|
|
|
|
def update_user_disabled(
|
|
conn: connection,
|
|
username: str,
|
|
disabled: bool
|
|
):
|
|
# if disabled = True => user is disabled
|
|
with conn.cursor() as cur:
|
|
cur.execute(
|
|
"""
|
|
update picrinth.users
|
|
set disabled = %s
|
|
where username = %s
|
|
""",
|
|
(disabled, username),
|
|
)
|
|
conn.commit()
|
|
return cur.rowcount > 0
|
|
|
|
def update_user_role(
|
|
conn: connection,
|
|
username: str,
|
|
role: str
|
|
):
|
|
with conn.cursor() as cur:
|
|
cur.execute(
|
|
"""
|
|
update picrinth.users
|
|
set role = %s
|
|
where username = %s
|
|
""",
|
|
(role, username),
|
|
)
|
|
conn.commit()
|
|
return cur.rowcount > 0
|
|
|
|
|
|
def update_user_username(
|
|
conn: connection,
|
|
username: str,
|
|
newUsername: str
|
|
):
|
|
with conn.cursor() as cur:
|
|
cur.execute(
|
|
"""
|
|
update picrinth.users
|
|
set username = %s
|
|
where username = %s;
|
|
""",
|
|
(newUsername, username),
|
|
)
|
|
conn.commit()
|
|
return cur.rowcount > 0
|
|
|
|
def update_user_password(
|
|
conn: connection,
|
|
username: str,
|
|
password: str
|
|
):
|
|
with conn.cursor() as cur:
|
|
cur.execute(
|
|
"""
|
|
update picrinth.users
|
|
set password = %s
|
|
where username = %s
|
|
""",
|
|
(password, username),
|
|
)
|
|
conn.commit()
|
|
return cur.rowcount > 0
|
|
|
|
|
|
def update_user_last_seen(
|
|
conn: connection,
|
|
username: str
|
|
):
|
|
with conn.cursor() as cur:
|
|
cur.execute(
|
|
"""
|
|
update picrinth.users
|
|
set last_seen_at = now()
|
|
where username = %s
|
|
""",
|
|
(username,),
|
|
)
|
|
conn.commit()
|
|
return cur.rowcount > 0
|
|
|
|
|
|
# user receiving
|
|
|
|
def get_user(
|
|
conn: connection,
|
|
username: str
|
|
):
|
|
with conn.cursor(cursor_factory=psycopg2.extras.DictCursor) as cur:
|
|
cur.execute(
|
|
"""
|
|
select username, password, role,
|
|
disabled, last_seen_at, created_at
|
|
from picrinth.users
|
|
where username = %s
|
|
""",
|
|
(username,),
|
|
)
|
|
return cur.fetchone()
|
|
|
|
def get_user_password(
|
|
conn: connection,
|
|
username: str
|
|
):
|
|
with conn.cursor() as cur:
|
|
cur.execute(
|
|
"""
|
|
select password
|
|
from picrinth.users
|
|
where username = %s
|
|
""",
|
|
(username,),
|
|
)
|
|
result = cur.fetchone()
|
|
if result is None:
|
|
return None
|
|
return result[0] # type: ignore
|