From 3e12c54b1d5b2d85410e69e6c048cf9e646eb106 Mon Sep 17 00:00:00 2001 From: Jennifer Taylor Date: Thu, 13 Aug 2026 01:00:07 +0000 Subject: [PATCH] Fix type hints after newer MyPy update. --- bemani/common/decorators.py | 2 +- bemani/data/mysql/api.py | 11 ++++++----- bemani/data/mysql/base.py | 6 +++--- bemani/data/mysql/game.py | 8 ++++---- bemani/data/mysql/lobby.py | 9 +++++---- bemani/data/mysql/machine.py | 14 +++++++------- bemani/data/mysql/music.py | 10 +++++----- bemani/data/mysql/network.py | 4 ++-- bemani/data/mysql/user.py | 30 +++++++++++++++--------------- bemani/utils/read.py | 10 +++++----- 10 files changed, 53 insertions(+), 51 deletions(-) diff --git a/bemani/common/decorators.py b/bemani/common/decorators.py index 4655d75..2e9b805 100644 --- a/bemani/common/decorators.py +++ b/bemani/common/decorators.py @@ -1,6 +1,6 @@ from typing import TypeVar -T = TypeVar('T') +T = TypeVar("T") def debugonly(func: T) -> T: diff --git a/bemani/data/mysql/api.py b/bemani/data/mysql/api.py index 4f60893..d8a9fe1 100644 --- a/bemani/data/mysql/api.py +++ b/bemani/data/mysql/api.py @@ -1,7 +1,8 @@ import uuid from sqlalchemy import Table, Column +from sqlalchemy.engine import RowMapping from sqlalchemy.types import String, Integer -from typing import Any, Dict, List, Optional +from typing import List, Optional from bemani.common import Time from bemani.data.mysql.base import BaseData, metadata @@ -70,7 +71,7 @@ class APIData(APIProviderInterface, BaseData): """ sql = "SELECT count(*) AS count FROM client WHERE token = :token" cursor = self.execute(sql, {"token": token}) - return cursor.mappings().fetchone()["count"] == 1 # type: ignore + return cursor.mappings().fetchone()["count"] == 1 def create_client(self, name: str) -> int: """ @@ -109,7 +110,7 @@ class APIData(APIProviderInterface, BaseData): # Couldn't find an entry with this ID return None - result = cursor.mappings().fetchone() # type: ignore + result = cursor.mappings().fetchone() return Client( clientid, result["timestamp"], @@ -145,7 +146,7 @@ class APIData(APIProviderInterface, BaseData): A list of Server objects sorted by add time. """ - def format_result(result: Dict[str, Any]) -> Server: + def format_result(result: RowMapping) -> Server: allow_stats = (result["config"] & 0x1) == 0 allow_scores = (result["config"] & 0x2) == 0 return Server( @@ -199,7 +200,7 @@ class APIData(APIProviderInterface, BaseData): # Couldn't find an entry with this ID return None - result = cursor.mappings().fetchone() # type: ignore + result = cursor.mappings().fetchone() allow_stats = (result["config"] & 0x1) == 0 allow_scores = (result["config"] & 0x2) == 0 return Server( diff --git a/bemani/data/mysql/base.py b/bemani/data/mysql/base.py index 2be91f9..9f9de76 100644 --- a/bemani/data/mysql/base.py +++ b/bemani/data/mysql/base.py @@ -1,6 +1,6 @@ import json import random -from typing import Final, Dict, Any, Optional +from typing import Final, Dict, Any, Optional, cast from bemani.common import Time from bemani.data.config import Config @@ -86,7 +86,7 @@ class BaseData: params if params is not None else {}, ) self.__conn.commit() - return result + return cast(CursorResult, result) def serialize(self, data: Dict[str, Any]) -> str: """ @@ -142,7 +142,7 @@ class BaseData: # Couldn't find a user with this session return None - result = cursor.mappings().fetchone() # type: ignore + result = cursor.mappings().fetchone() return result["id"] def _create_session(self, opid: int, optype: str, expiration: int = (30 * 86400)) -> str: diff --git a/bemani/data/mysql/game.py b/bemani/data/mysql/game.py index f7bb29b..04b4863 100644 --- a/bemani/data/mysql/game.py +++ b/bemani/data/mysql/game.py @@ -96,7 +96,7 @@ class GameData(BaseData): # Settings doesn't exist return None - result = cursor.mappings().fetchone() # type: ignore + result = cursor.mappings().fetchone() return ValidatedDict(self.deserialize(result["data"])) def put_settings(self, game: GameConstants, userid: UserID, settings: Dict[str, Any]) -> None: @@ -158,7 +158,7 @@ class GameData(BaseData): # score doesn't exist return None - result = cursor.mappings().fetchone() # type: ignore + result = cursor.mappings().fetchone() return ValidatedDict(self.deserialize(result["data"])) def get_achievements(self, game: GameConstants, userid: UserID) -> List[Achievement]: @@ -251,7 +251,7 @@ class GameData(BaseData): # setting doesn't exist return None - result = cursor.mappings().fetchone() # type: ignore + result = cursor.mappings().fetchone() retval = ValidatedDict(self.deserialize(result["data"])) retval["start_time"] = result["start_time"] retval["end_time"] = result["end_time"] @@ -387,7 +387,7 @@ class GameData(BaseData): # entry doesn't exist return None - result = cursor.mappings().fetchone() # type: ignore + result = cursor.mappings().fetchone() return ValidatedDict(self.deserialize(result["data"])) def get_items(self, game: GameConstants, version: int) -> List[Item]: diff --git a/bemani/data/mysql/lobby.py b/bemani/data/mysql/lobby.py index d75434f..f2d4e66 100644 --- a/bemani/data/mysql/lobby.py +++ b/bemani/data/mysql/lobby.py @@ -1,6 +1,7 @@ import copy from sqlalchemy import Table, Column, UniqueConstraint +from sqlalchemy.engine import RowMapping from sqlalchemy.types import String, Integer, JSON from sqlalchemy.dialects.mysql import BIGINT as BigInteger from typing import Optional, Dict, List, Tuple, Any @@ -82,7 +83,7 @@ class LobbyData(BaseData): # Settings doesn't exist return None - result = cursor.mappings().fetchone() # type: ignore + result = cursor.mappings().fetchone() data = ValidatedDict(self.deserialize(result["data"])) data["id"] = result["id"] data["time"] = result["time"] @@ -113,7 +114,7 @@ class LobbyData(BaseData): }, ) - def format_result(result: Dict[str, Any]) -> ValidatedDict: + def format_result(result: RowMapping) -> ValidatedDict: data = ValidatedDict(self.deserialize(result["data"])) data["id"] = result["id"] data["time"] = result["time"] @@ -214,7 +215,7 @@ class LobbyData(BaseData): # Settings doesn't exist return None - result = cursor.mappings().fetchone() # type: ignore + result = cursor.mappings().fetchone() data = ValidatedDict(self.deserialize(result["data"])) data["id"] = result["id"] data["time"] = result["time"] @@ -246,7 +247,7 @@ class LobbyData(BaseData): }, ) - def format_result(result: Dict[str, Any]) -> ValidatedDict: + def format_result(result: RowMapping) -> ValidatedDict: data = ValidatedDict(self.deserialize(result["data"])) data["id"] = result["id"] data["time"] = result["time"] diff --git a/bemani/data/mysql/machine.py b/bemani/data/mysql/machine.py index 199ac20..aafd589 100644 --- a/bemani/data/mysql/machine.py +++ b/bemani/data/mysql/machine.py @@ -101,7 +101,7 @@ class MachineData(BaseData): # Machine doesn't exist return None - result = cursor.mappings().fetchone() # type: ignore + result = cursor.mappings().fetchone() return result["pcbid"] def from_machine_id(self, machine_id: int) -> Optional[str]: @@ -121,7 +121,7 @@ class MachineData(BaseData): # Machine doesn't exist return None - result = cursor.mappings().fetchone() # type: ignore + result = cursor.mappings().fetchone() return result["pcbid"] def from_userid(self, userid: UserID) -> List[ArcadeID]: @@ -176,7 +176,7 @@ class MachineData(BaseData): # Machine doesn't exist return None - result = cursor.mappings().fetchone() # type: ignore + result = cursor.mappings().fetchone() return Machine( result["id"], pcbid, @@ -282,7 +282,7 @@ class MachineData(BaseData): port = None else: # Grab highest port - result = cursor.mappings().fetchone() # type: ignore + result = cursor.mappings().fetchone() port = result["port"] if port is not None: port = port + 1 @@ -355,7 +355,7 @@ class MachineData(BaseData): ) if cursor.rowcount != 1: raise ArcadeCreationException("Failed to create arcade!") - arcadeid = cursor.lastrowid + arcadeid = ArcadeID(cursor.lastrowid) for owner in owners: sql = """ INSERT INTO arcade_owner (userid, arcadeid) @@ -386,7 +386,7 @@ class MachineData(BaseData): # Arcade doesn't exist return None - result = cursor.mappings().fetchone() # type: ignore + result = cursor.mappings().fetchone() sql = "SELECT userid FROM arcade_owner WHERE arcadeid = :id" cursor = self.execute(sql, {"id": arcadeid}) @@ -515,7 +515,7 @@ class MachineData(BaseData): # Settings doesn't exist return None - result = cursor.mappings().fetchone() # type: ignore + result = cursor.mappings().fetchone() return ValidatedDict(self.deserialize(result["data"])) def put_settings( diff --git a/bemani/data/mysql/music.py b/bemani/data/mysql/music.py index 80444cc..33c4e4d 100644 --- a/bemani/data/mysql/music.py +++ b/bemani/data/mysql/music.py @@ -107,7 +107,7 @@ class MusicData(BaseData): if cursor.rowcount != 1: # music doesn't exist raise Exception(f"Song {songid} chart {songchart} doesn't exist for game {game} version {version}") - result = cursor.mappings().fetchone() # type: ignore + result = cursor.mappings().fetchone() return result["id"] def put_score( @@ -296,7 +296,7 @@ class MusicData(BaseData): # score doesn't exist return None - result = cursor.mappings().fetchone() # type: ignore + result = cursor.mappings().fetchone() return Score( result["scorekey"], result["songid"], @@ -356,7 +356,7 @@ class MusicData(BaseData): # score doesn't exist return None - result = cursor.mappings().fetchone() # type: ignore + result = cursor.mappings().fetchone() return ( UserID(result["userid"]), Score( @@ -597,7 +597,7 @@ class MusicData(BaseData): if cursor.rowcount != 1: # music doesn't exist return None - result = cursor.mappings().fetchone() # type: ignore + result = cursor.mappings().fetchone() return Song( game, version, @@ -908,7 +908,7 @@ class MusicData(BaseData): # score doesn't exist return None - result = cursor.mappings().fetchone() # type: ignore + result = cursor.mappings().fetchone() return ( UserID(result["userid"]), Attempt( diff --git a/bemani/data/mysql/network.py b/bemani/data/mysql/network.py index 6152e5f..27a8f28 100644 --- a/bemani/data/mysql/network.py +++ b/bemani/data/mysql/network.py @@ -107,7 +107,7 @@ class NetworkData(BaseData): # Couldn't find an entry with this ID return None - result = cursor.mappings().fetchone() # type: ignore + result = cursor.mappings().fetchone() return News( newsid, result["timestamp"], @@ -181,7 +181,7 @@ class NetworkData(BaseData): # No scheduled work was registered, so time to get going! return True - result = cursor.mappings().fetchone() # type: ignore + result = cursor.mappings().fetchone() if schedule == "daily": # Just look at the day and year, make sure it matches diff --git a/bemani/data/mysql/user.py b/bemani/data/mysql/user.py index 1bb1186..92f4709 100644 --- a/bemani/data/mysql/user.py +++ b/bemani/data/mysql/user.py @@ -193,7 +193,7 @@ class UserData(BaseData): # Couldn't find a user with this card return None - result = cursor.mappings().fetchone() # type: ignore + result = cursor.mappings().fetchone() return UserID(result["userid"]) def from_username(self, username: str) -> Optional[UserID]: @@ -212,7 +212,7 @@ class UserData(BaseData): # Couldn't find this username return None - result = cursor.mappings().fetchone() # type: ignore + result = cursor.mappings().fetchone() return UserID(result["id"]) def from_refid(self, game: GameConstants, version: int, refid: str) -> Optional[UserID]: @@ -237,7 +237,7 @@ class UserData(BaseData): # Couldn't find a user with this refid return None - result = cursor.mappings().fetchone() # type: ignore + result = cursor.mappings().fetchone() return UserID(result["userid"]) def from_extid(self, game: GameConstants, extid: int) -> Optional[UserID]: @@ -261,7 +261,7 @@ class UserData(BaseData): # Couldn't find a user with this refid return None - result = cursor.mappings().fetchone() # type: ignore + result = cursor.mappings().fetchone() return UserID(result["userid"]) def from_session(self, session: str) -> Optional[UserID]: @@ -300,7 +300,7 @@ class UserData(BaseData): # User doesn't exist, but we have a reference? return None - result = cursor.mappings().fetchone() # type: ignore + result = cursor.mappings().fetchone() return User(userid, result["username"], result["email"], result["admin"] == 1) def get_all_users(self) -> List[User]: @@ -440,7 +440,7 @@ class UserData(BaseData): # User doesn't exist, but we have a reference? return False - result = cursor.mappings().fetchone() # type: ignore + result = cursor.mappings().fetchone() return pin == result["pin"] def update_pin(self, userid: UserID, pin: str) -> None: @@ -471,7 +471,7 @@ class UserData(BaseData): # User doesn't exist, but we have a reference? return False - result = cursor.mappings().fetchone() # type: ignore + result = cursor.mappings().fetchone() passhash = result["password"] try: @@ -520,7 +520,7 @@ class UserData(BaseData): # Profile doesn't exist return None - result = cursor.mappings().fetchone() # type: ignore + result = cursor.mappings().fetchone() return Profile( game, version, @@ -849,7 +849,7 @@ class UserData(BaseData): # score doesn't exist return None - result = cursor.mappings().fetchone() # type: ignore + result = cursor.mappings().fetchone() return ValidatedDict(self.deserialize(result["data"])) def get_achievements(self, game: GameConstants, version: int, userid: UserID) -> List[Achievement]: @@ -1115,7 +1115,7 @@ class UserData(BaseData): # score doesn't exist return None - result = cursor.mappings().fetchone() # type: ignore + result = cursor.mappings().fetchone() return ValidatedDict(self.deserialize(result["data"])) def get_links(self, game: GameConstants, version: int, userid: UserID) -> List[Link]: @@ -1237,7 +1237,7 @@ class UserData(BaseData): sql = "SELECT balance FROM balance WHERE userid = :userid AND arcadeid = :arcadeid" cursor = self.execute(sql, {"userid": userid, "arcadeid": arcadeid}) if cursor.rowcount == 1: - result = cursor.mappings().fetchone() # type: ignore + result = cursor.mappings().fetchone() return result["balance"] else: return 0 @@ -1283,7 +1283,7 @@ class UserData(BaseData): sql = "SELECT refid FROM refid WHERE userid = :userid AND game = :game AND version = :version" cursor = self.execute(sql, {"userid": userid, "game": game.value, "version": version}) if cursor.rowcount == 1: - result = cursor.mappings().fetchone() # type: ignore + result = cursor.mappings().fetchone() return result["refid"] else: return self.create_refid(game, version, userid) @@ -1306,7 +1306,7 @@ class UserData(BaseData): sql = "SELECT extid FROM extid WHERE userid = :userid AND game = :game" cursor = self.execute(sql, {"userid": userid, "game": game.value}) if cursor.rowcount == 1: - result = cursor.mappings().fetchone() # type: ignore + result = cursor.mappings().fetchone() return result["extid"] else: return None @@ -1412,7 +1412,7 @@ class UserData(BaseData): sql = "SELECT refid FROM refid WHERE userid = :userid AND game = :game AND version = :version" cursor = self.execute(sql, {"userid": userid, "game": game.value, "version": version}) if cursor.rowcount == 1: - result = cursor.mappings().fetchone() # type: ignore + result = cursor.mappings().fetchone() return result["refid"] # Shouldn't be possible, but here we are raise AccountCreationException("Failed to recover lost race refid!") @@ -1459,4 +1459,4 @@ class UserData(BaseData): self.execute(sql, {"newid": userid, "oldid": oldid}) # Finally, return the user ID - return userid + return UserID(userid) diff --git a/bemani/utils/read.py b/bemani/utils/read.py index 7cfc226..a6bbef7 100644 --- a/bemani/utils/read.py +++ b/bemani/utils/read.py @@ -14,7 +14,7 @@ from sqlalchemy.engine import CursorResult from sqlalchemy.orm import scoped_session, sessionmaker from sqlalchemy.sql import text from sqlalchemy.exc import IntegrityError -from typing import Any, Callable, Dict, List, Optional, Tuple +from typing import Any, Callable, Dict, List, Optional, Tuple, cast from bemani.common import ( GameConstants, @@ -97,7 +97,7 @@ class ImportBase: ]: if write_statement in sql.lower(): raise Exception("Read-only mode is active!") - return self.__conn.execute(text(sql), params if params is not None else {}) + return cast(CursorResult, self.__conn.execute(text(sql), params if params is not None else {})) def remote_music(self, server: str, token: str) -> GlobalMusicData: api = ReadAPI(server, token) @@ -111,7 +111,7 @@ class ImportBase: def get_next_music_id(self) -> int: cursor = self.execute("SELECT MAX(id) AS next_id FROM `music`") - result = cursor.mappings().fetchone() # type: ignore + result = cursor.mappings().fetchone() try: return result["next_id"] + 1 except TypeError: @@ -139,7 +139,7 @@ class ImportBase: }, ) if cursor.rowcount != 0: - result = cursor.mappings().fetchone() # type: ignore + result = cursor.mappings().fetchone() return result["id"] else: return None @@ -184,7 +184,7 @@ class ImportBase: }, ) if cursor.rowcount != 0: - result = cursor.mappings().fetchone() # type: ignore + result = cursor.mappings().fetchone() return result["id"] else: return None