50 lines
2.1 KiB
Python
50 lines
2.1 KiB
Python
"""Доступ к данным пользователей."""
|
||
|
||
from __future__ import annotations
|
||
|
||
from typing import Any
|
||
|
||
from common.db.pool import get_pool
|
||
|
||
|
||
async def add_user(id_user: int, username: str, first_name: str, last_name: str) -> None:
|
||
"""Создать/обновить пользователя и добавить запись о сообщении."""
|
||
async with get_pool().connection() as conn, conn.cursor() as cur:
|
||
await cur.execute(
|
||
"CALL bot.add_user(%s, %s, %s, %s)",
|
||
(id_user, username, first_name, last_name),
|
||
)
|
||
|
||
|
||
async def add_nick(id_user: int, nick: str) -> None:
|
||
"""Установить ник пользователя."""
|
||
async with get_pool().connection() as conn, conn.cursor() as cur:
|
||
await cur.execute("CALL bot.add_nick(%s, %s)", (id_user, nick))
|
||
|
||
|
||
async def get_nick_list() -> list[dict[str, Any]]:
|
||
"""Список всех игроков (username, nick), отсортированный по нику."""
|
||
async with get_pool().connection() as conn, conn.cursor() as cur:
|
||
await cur.execute("SELECT username, nick FROM bot.get_nick_list()")
|
||
return await cur.fetchall()
|
||
|
||
|
||
async def get_nick_list_subscribe() -> list[dict[str, Any]]:
|
||
"""Список подписанных на рассылки (username, nick)."""
|
||
async with get_pool().connection() as conn, conn.cursor() as cur:
|
||
await cur.execute("SELECT username, nick FROM bot.get_nick_list_subscribe()")
|
||
return await cur.fetchall()
|
||
|
||
|
||
async def get_info_by_id(id_user: int) -> dict[str, Any] | None:
|
||
"""Сводная информация о пользователе."""
|
||
async with get_pool().connection() as conn, conn.cursor() as cur:
|
||
await cur.execute("SELECT * FROM bot.get_info_by_id(%s)", (id_user,))
|
||
return await cur.fetchone()
|
||
|
||
|
||
async def del_user_by_id(id_user: int) -> None:
|
||
"""Удалить пользователя (при выходе из чата)."""
|
||
async with get_pool().connection() as conn, conn.cursor() as cur:
|
||
await cur.execute("CALL bot.del_user_by_id(%s)", (id_user,))
|