118 lines
2.9 KiB
Python
118 lines
2.9 KiB
Python
from typing import List
|
|
|
|
from src.internals.database.database import query_db, query_one_db, query_rowcount_db
|
|
from src.lib.notification import send_notifications, TDNotificationInit
|
|
from src.lib.pagination import TDPagination
|
|
from src.types.account import Account, AccountRoleChange, NotificationTypes, visible_roles
|
|
|
|
|
|
def get_account(account_id: str) -> Account:
|
|
account = query_one_db(
|
|
"""
|
|
SELECT id, username, created_at, role
|
|
FROM account
|
|
WHERE id = %s
|
|
""",
|
|
(account_id,),
|
|
)
|
|
account = Account.init_from_dict(account)
|
|
|
|
return account
|
|
|
|
|
|
def count_accounts(role: str | None = None, name: str | None = None) -> int:
|
|
roles = [role] if role else visible_roles
|
|
params = dict(
|
|
roles=roles,
|
|
name=f"%%{name}%%" if name else None
|
|
)
|
|
query = """
|
|
SELECT
|
|
COUNT(*)
|
|
FROM
|
|
account
|
|
WHERE
|
|
role = ANY(%(roles)s::varchar[])
|
|
AND
|
|
(
|
|
%(name)s::varchar IS NULL
|
|
OR
|
|
username LIKE %(name)s::varchar
|
|
)
|
|
"""
|
|
result = query_one_db(
|
|
query,
|
|
params,
|
|
)
|
|
count = result["count"] if result else 0
|
|
|
|
return count
|
|
|
|
|
|
def get_accounts(pagination: TDPagination, role: str | None = None, name: str | None = None) -> List[Account]:
|
|
roles = [role] if role else visible_roles
|
|
params = dict(
|
|
roles=roles,
|
|
name=f"%%{name}%%" if name else None,
|
|
offset=pagination["offset"],
|
|
limit=pagination["limit"]
|
|
)
|
|
query = """
|
|
SELECT
|
|
id,
|
|
username,
|
|
created_at,
|
|
role
|
|
FROM
|
|
account
|
|
WHERE
|
|
role = ANY(%(roles)s::varchar[])
|
|
AND
|
|
(
|
|
%(name)s::varchar IS NULL
|
|
OR
|
|
username LIKE %(name)s::varchar
|
|
)
|
|
ORDER BY
|
|
id ASC
|
|
OFFSET %(offset)s
|
|
LIMIT %(limit)s
|
|
"""
|
|
accounts = query_db(
|
|
query,
|
|
params,
|
|
)
|
|
acc_list = [Account.init_from_dict(acc) for acc in accounts]
|
|
|
|
return acc_list
|
|
|
|
|
|
def change_account_role(account_ids: List[str], extra_info: AccountRoleChange):
|
|
change_role_query = """
|
|
UPDATE
|
|
account
|
|
SET
|
|
role = %s
|
|
WHERE
|
|
id = ANY (%s)
|
|
"""
|
|
query_rowcount_db(change_role_query, (extra_info["new_role"], account_ids))
|
|
|
|
notification_inits: list[TDNotificationInit] = [
|
|
TDNotificationInit(
|
|
account_id=account_id,
|
|
type=NotificationTypes.ACCOUNT_ROLE_CHANGE,
|
|
# resort to this retardation because
|
|
# psycopg chokes json'ing typed dicts
|
|
extra_info=dict(
|
|
old_role=extra_info["old_role"],
|
|
new_role=extra_info["new_role"]
|
|
)
|
|
)
|
|
for account_id
|
|
in account_ids
|
|
]
|
|
send_notifications(notification_inits)
|
|
|
|
return True
|