Leahnaya/bemaniutils
1
0
Fork 0
mirror of https://github.com/DragonMinded/bemaniutils.git synced 2026-08-19 17:06:41 -05:00
Code Issues Actions Packages Projects Releases Wiki Activity

Fix type hints after newer MyPy update.

Browse Source
This commit is contained in:
Jennifer Taylor
2026-08-13 01:00:07 +00:00
parent 13bb6da26a
commit 3e12c54b1d
10 changed files with 53 additions and 51 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
bemani/common/decorators.py
View File

@@ -1,6 +1,6 @@
from typing import TypeVar
T = TypeVar('T')
T = TypeVar("T")
def debugonly(func: T) -> T:

11
bemani/data/mysql/api.py
View File

@@ -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(

6
bemani/data/mysql/base.py
View File

@@ -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:

8
bemani/data/mysql/game.py
View File

@@ -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]:

9
bemani/data/mysql/lobby.py
View File

@@ -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"]

14
bemani/data/mysql/machine.py
View File

@@ -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(

10
bemani/data/mysql/music.py
View File

@@ -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(

4
bemani/data/mysql/network.py
View File

@@ -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

30
bemani/data/mysql/user.py
View File

@@ -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)

10
bemani/utils/read.py
View File

@@ -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