From b2e273932fc035db9de69042d650a125249b8ffd Mon Sep 17 00:00:00 2001 From: Alonso <33896771+7mochi@users.noreply.github.com> Date: Sat, 23 Sep 2023 23:24:16 -0500 Subject: [PATCH] refactor: ingame_logins repository (#489) * feat: base ingame_logins repository * fix: rename InGameLogin to IngameLogin * fix: fix typing errors * fix: apply suggestions * fix: apply suggestions * type defs --------- Co-authored-by: cmyui --- app/api/domains/cho.py | 16 ++-- app/repositories/ingame_logins.py | 148 ++++++++++++++++++++++++++++++ 2 files changed, 154 insertions(+), 10 deletions(-) create mode 100644 app/repositories/ingame_logins.py diff --git a/app/api/domains/cho.py b/app/api/domains/cho.py index 54696f02..7db471b2 100644 --- a/app/api/domains/cho.py +++ b/app/api/domains/cho.py @@ -60,6 +60,7 @@ from app.packets import BanchoPacketReader from app.packets import BasePacket from app.packets import ClientPackets +from app.repositories import ingame_logins as logins_repo from app.repositories import players as players_repo from app.state import services from app.usecases.performance import ScoreParams @@ -697,16 +698,11 @@ async def login( """ login credentials verified """ - await db_conn.execute( - "INSERT INTO ingame_logins " - "(userid, ip, osu_ver, osu_stream, datetime) " - "VALUES (:id, :ip, :osu_ver, :osu_stream, NOW())", - { - "id": user_info["id"], - "ip": str(ip), - "osu_ver": osu_version.date, - "osu_stream": osu_version.stream, - }, + await logins_repo.create( + user_id=user_info["id"], + ip=str(ip), + osu_ver=osu_version.date, + osu_stream=osu_version.stream, ) await db_conn.execute( diff --git a/app/repositories/ingame_logins.py b/app/repositories/ingame_logins.py new file mode 100644 index 00000000..fa65045f --- /dev/null +++ b/app/repositories/ingame_logins.py @@ -0,0 +1,148 @@ +from __future__ import annotations + +import textwrap +from datetime import datetime +from typing import Any +from typing import cast +from typing import TypedDict + +import app.state.services +from app._typing import _UnsetSentinel +from app._typing import UNSET + +# +--------------+------------------------+------+-----+---------+-------+ +# | Field | Type | Null | Key | Default | Extra | +# +--------------+------------------------+------+-----+---------+-------+ +# | id | int | NO | PRI | NULL | | +# | userid | int | NO | | NULL | | +# | ip | varchar(45) | NO | | NULL | | +# | osu_ver | date | NO | | NULL | | +# | osu_stream | varchar(11) | NO | | NULL | | +# | datetime | datetime | NO | | NULL | | +# +--------------+------------------------+------+-----+---------+-------+ + +READ_PARAMS = textwrap.dedent( + """\ + id, userid, ip, osu_ver, osu_stream, datetime + """, +) + + +class IngameLogin(TypedDict): + id: int + userid: str + ip: str + osu_ver: str + osu_stream: str + datetime: datetime + + +class InGameLoginUpdateFields(TypedDict, total=False): + userid: str + ip: str + osu_ver: str + osu_stream: str + + +async def create( + user_id: int, + ip: str, + osu_ver: str, + osu_stream: str, +) -> IngameLogin: + """Create a new login entry in the database.""" + query = f"""\ + INSERT INTO ingame_logins (userid, ip, osu_ver, osu_stream, datetime) + VALUES (:userid, :ip, :osu_ver, :osu_stream, NOW()) + """ + params: dict[str, Any] = { + "userid": user_id, + "ip": ip, + "osu_ver": osu_ver, + "osu_stream": osu_stream, + } + rec_id = await app.state.services.database.execute(query, params) + + query = f"""\ + SELECT {READ_PARAMS} + FROM ingame_logins + WHERE id = :id + """ + params = { + "id": rec_id, + } + ingame_login = await app.state.services.database.fetch_one(query, params) + + assert ingame_login is not None + return cast(IngameLogin, dict(ingame_login._mapping)) + + +async def fetch_one(id: int) -> IngameLogin | None: + """Fetch a login entry from the database.""" + query = f"""\ + SELECT {READ_PARAMS} + FROM ingame_logins + WHERE id = :id + """ + params: dict[str, Any] = { + "id": id, + } + ingame_login = await app.state.services.database.fetch_one(query, params) + + return cast(IngameLogin, ingame_login) if ingame_login is not None else None + + +async def fetch_count( + user_id: int | None = None, + ip: str | None = None, +) -> int: + """Fetch the number of logins in the database.""" + query = """\ + SELECT COUNT(*) AS count + FROM ingame_logins + WHERE userid = COALESCE(:userid, userid) + AND ip = COALESCE(:ip, ip) + """ + params: dict[str, Any] = { + "userid": user_id, + "ip": ip, + } + rec = await app.state.services.database.fetch_one(query, params) + assert rec is not None + return cast(int, rec._mapping["count"]) + + +async def fetch_many( + user_id: int | None = None, + ip: str | None = None, + osu_ver: str | None = None, + osu_stream: str | None = None, + page: int | None = None, + page_size: int | None = None, +) -> list[IngameLogin]: + """Fetch a list of logins from the database.""" + query = f"""\ + SELECT {READ_PARAMS} + FROM ingame_logins + WHERE userid = COALESCE(:userid, userid) + AND ip = COALESCE(:ip, ip) + AND osu_ver = COALESCE(:osu_ver, osu_ver) + AND osu_stream = COALESCE(:osu_stream, osu_stream) + """ + params: dict[str, Any] = { + "userid": user_id, + "ip": ip, + "osu_ver": osu_ver, + "osu_stream": osu_stream, + } + + if page is not None and page_size is not None: + query += """\ + LIMIT :limit + OFFSET :offset + """ + params["limit"] = page_size + params["offset"] = (page - 1) * page_size + + ingame_logins = await app.state.services.database.fetch_all(query, params) + return cast(list[IngameLogin], ingame_logins)