Hay1tsme/artemis
30
52
Fork 48
Code Issues 5 Pull Requests 1 Packages Projects Releases Wiki Activity

[core] Address SQLAlchemy deprecations

Browse Source
This commit is contained in:
beerpsi
2024-11-18 10:49:41 +07:00
parent 58a5177a30
commit c11aae58a6
6 changed files with 156 additions and 79 deletions
Download Patch File Download Diff File Expand all files Collapse all files

80
core/data/schema/arcade.py
View File

@ -3,7 +3,7 @@ from typing import List, Optional
from sqlalchemy import Column, Table, and_, or_
from sqlalchemy.dialects.mysql import insert
from sqlalchemy.engine import Row
from sqlalchemy.engine import Row, RowMapping
from sqlalchemy.sql import func, select
from sqlalchemy.sql.schema import ForeignKey, PrimaryKeyConstraint
from sqlalchemy.types import JSON, Boolean, Integer, String
@ -69,21 +69,23 @@ arcade_owner: Table = Table(
class ArcadeData(BaseData):
async def get_machine(self, serial: Optional[str] = None, id: Optional[int] = None) -> Optional[Row]:
async def get_machine(
self, serial: Optional[str] = None, id: Optional[int] = None
) -> Optional[RowMapping]:
if serial is not None:
serial = serial.replace("-", "")
if len(serial) == 11:
sql = machine.select(machine.c.serial.like(f"{serial}%"))
sql = machine.select().where(machine.c.serial.like(f"{serial}%"))
elif len(serial) == 15:
sql = machine.select(machine.c.serial == serial)
sql = machine.select().where(machine.c.serial == serial)
else:
self.logger.error(f"{__name__ }: Malformed serial {serial}")
return None
elif id is not None:
sql = machine.select(machine.c.id == id)
sql = machine.select().where(machine.c.id == id)
else:
self.logger.error(f"{__name__ }: Need either serial or ID to look up!")
@ -92,7 +94,7 @@ class ArcadeData(BaseData):
result = await self.execute(sql)
if result is None:
return None
return result.fetchone()
return result.mappings().fetchone()
async def create_machine(
self,
@ -117,7 +119,7 @@ class ArcadeData(BaseData):
async def set_machine_serial(self, machine_id: int, serial: str) -> None:
result = await self.execute(
machine.update(machine.c.id == machine_id).values(keychip=serial)
machine.update().where(machine.c.id == machine_id).values(keychip=serial)
)
if result is None:
self.logger.error(
@ -127,26 +129,26 @@ class ArcadeData(BaseData):
async def set_machine_boardid(self, machine_id: int, boardid: str) -> None:
result = await self.execute(
machine.update(machine.c.id == machine_id).values(board=boardid)
machine.update().where(machine.c.id == machine_id).values(board=boardid)
)
if result is None:
self.logger.error(
f"Failed to update board id for machine {machine_id} -> {boardid}"
)
async def get_arcade(self, id: int) -> Optional[Row]:
sql = arcade.select(arcade.c.id == id)
async def get_arcade(self, id: int) -> Optional[RowMapping]:
sql = arcade.select().where(arcade.c.id == id)
result = await self.execute(sql)
if result is None:
return None
return result.fetchone()
async def get_arcade_machines(self, id: int) -> Optional[List[Row]]:
sql = machine.select(machine.c.arcade == id)
return result.mappings().fetchone()
async def get_arcade_machines(self, id: int) -> Optional[List[RowMapping]]:
sql = machine.select().where(machine.c.arcade == id)
result = await self.execute(sql)
if result is None:
return None
return result.fetchall()
return result.mappings().fetchall()
async def create_arcade(
self,
@ -177,14 +179,22 @@ class ArcadeData(BaseData):
return result.lastrowid
async def get_arcades_managed_by_user(self, user_id: int) -> Optional[List[Row]]:
sql = select(arcade).join(arcade_owner, arcade_owner.c.arcade == arcade.c.id).where(arcade_owner.c.user == user_id)
sql = (
select(arcade)
.join(arcade_owner, arcade_owner.c.arcade == arcade.c.id)
.where(arcade_owner.c.user == user_id)
)
result = await self.execute(sql)
if result is None:
return False
return result.fetchall()
async def get_manager_permissions(self, user_id: int, arcade_id: int) -> Optional[int]:
sql = select(arcade_owner.c.permissions).where(and_(arcade_owner.c.user == user_id, arcade_owner.c.arcade == arcade_id))
async def get_manager_permissions(
self, user_id: int, arcade_id: int
) -> Optional[int]:
sql = select(arcade_owner.c.permissions).where(
and_(arcade_owner.c.user == user_id, arcade_owner.c.arcade == arcade_id)
)
result = await self.execute(sql)
if result is None:
return False
@ -207,7 +217,9 @@ class ArcadeData(BaseData):
return result.lastrowid
async def get_arcade_by_name(self, name: str) -> Optional[List[Row]]:
sql = arcade.select(or_(arcade.c.name.like(f"%{name}%"), arcade.c.nickname.like(f"%{name}%")))
sql = arcade.select().where(
or_(arcade.c.name.like(f"%{name}%"), arcade.c.nickname.like(f"%{name}%"))
)
result = await self.execute(sql)
if result is None:
return None
@ -219,25 +231,38 @@ class ArcadeData(BaseData):
if result is None:
return None
return result.fetchall()
async def get_num_generated_keychips(self) -> Optional[int]:
result = await self.execute(select(func.count("serial LIKE 'A69A%'")).select_from(machine))
result = await self.execute(
select(func.count("serial LIKE 'A69A%'")).select_from(machine)
)
if result:
return result.fetchone()['count_1']
return result.mappings().fetchone()["count_1"]
self.logger.error("Failed to count machine serials that start with A69A!")
def format_serial(
self, platform_code: str, platform_rev: int, serial_letter: str, serial_num: int, append: int, dash: bool = False
self,
platform_code: str,
platform_rev: int,
serial_letter: str,
serial_num: int,
append: int,
dash: bool = False,
) -> str:
return f"{platform_code}{'-' if dash else ''}{platform_rev:02d}{serial_letter}{serial_num:04d}{append:04d}"
def validate_keychip_format(self, serial: str) -> bool:
# For the 2nd letter, E and X are the only "real" values that have been observed (A is used for generated keychips)
if re.fullmatch(r"^A[0-9]{2}[A-Z][-]?[0-9]{2}[A-HJ-NP-Z][0-9]{4}([0-9]{4})?$", serial) is None:
if (
re.fullmatch(
r"^A[0-9]{2}[A-Z][-]?[0-9]{2}[A-HJ-NP-Z][0-9]{4}([0-9]{4})?$", serial
)
is None
):
return False
return True
# Thanks bottersnike!
def get_keychip_suffix(self, year: int, month: int) -> str:
assert year > 1957
@ -252,7 +277,6 @@ class ArcadeData(BaseData):
month = ((month - 1) + 9) % 12 # Offset so April=0
return f"{year:02}{month // 6:01}{month % 6 + 1:01}"
def parse_keychip_suffix(self, suffix: str) -> tuple[int, int]:
year = int(suffix[0:2])
half = int(suffix[2])