misc: Migrate to SQLAlchemy declarative models

This change applies the guided migration process recommended by SQLAlchemy [1], up to step 4, to have declarative ORM models that better support Python typing. The change was tested by running `alembic check`, which does not find any schema changes. Errors reported by `mypy` go down to 170, from the original 223 in the current `master` commit. [1] https://docs.sqlalchemy.org/en/20/changelog/whatsnew_20.html#migrating-an-existing-mapping
2026-02-18 23:42:07 +01:00 · 2024-06-23 20:40:04 -03:00
parent c3329012df
commit b99ededced
8 changed files with 151 additions and 151 deletions
--- a/backend/models/assets.py
+++ b/backend/models/assets.py
@@ -1,31 +1,33 @@
+from datetime import datetime
 from functools import cached_property
-from typing import Optional
+from typing import TYPE_CHECKING, Optional

 from models.base import BaseModel
-from sqlalchemy import BigInteger, Column, DateTime, ForeignKey, Integer, String, func
-from sqlalchemy.orm import relationship
+from sqlalchemy import BigInteger, DateTime, ForeignKey, String, func
+from sqlalchemy.orm import Mapped, mapped_column, relationship
+
+if TYPE_CHECKING:
+    from models.rom import Rom
+    from models.user import User


 class BaseAsset(BaseModel):
    __abstract__ = True

-    id = Column(Integer(), primary_key=True, autoincrement=True)
-    created_at = Column(
-        DateTime(timezone=True), server_default=func.now(), nullable=False
+    id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True)
+    created_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), server_default=func.now()
    )
-    updated_at = Column(
-        DateTime(timezone=True),
-        server_default=func.now(),
-        onupdate=func.now(),
-        nullable=False,
+    updated_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), server_default=func.now(), onupdate=func.now()
    )

-    file_name = Column(String(length=450), nullable=False)
-    file_name_no_tags = Column(String(length=450), nullable=False)
-    file_name_no_ext = Column(String(length=450), nullable=False)
-    file_extension = Column(String(length=100), nullable=False)
-    file_path = Column(String(length=1000), nullable=False)
-    file_size_bytes = Column(BigInteger(), default=0, nullable=False)
+    file_name: Mapped[str] = mapped_column(String(length=450))
+    file_name_no_tags: Mapped[str] = mapped_column(String(length=450))
+    file_name_no_ext: Mapped[str] = mapped_column(String(length=450))
+    file_extension: Mapped[str] = mapped_column(String(length=100))
+    file_path: Mapped[str] = mapped_column(String(length=1000))
+    file_size_bytes: Mapped[int] = mapped_column(BigInteger(), default=0)

    @cached_property
    def full_path(self) -> str:
@@ -39,22 +41,18 @@ class BaseAsset(BaseModel):
 class RomAsset(BaseAsset):
    __abstract__ = True

-    rom_id = Column(
-        Integer(), ForeignKey("roms.id", ondelete="CASCADE"), nullable=False
-    )
-    user_id = Column(
-        Integer(), ForeignKey("users.id", ondelete="CASCADE"), nullable=False
-    )
+    rom_id: Mapped[int] = mapped_column(ForeignKey("roms.id", ondelete="CASCADE"))
+    user_id: Mapped[int] = mapped_column(ForeignKey("users.id", ondelete="CASCADE"))


 class Save(RomAsset):
    __tablename__ = "saves"
    __table_args__ = {"extend_existing": True}

-    emulator = Column(String(length=50), nullable=True)
+    emulator: Mapped[str | None] = mapped_column(String(length=50))

-    rom = relationship("Rom", lazy="joined", back_populates="saves")
-    user = relationship("User", lazy="joined", back_populates="saves")
+    rom: Mapped["Rom"] = relationship(lazy="joined", back_populates="saves")
+    user: Mapped["User"] = relationship(lazy="joined", back_populates="saves")

    @cached_property
    def screenshot(self) -> Optional["Screenshot"]:
@@ -72,10 +70,10 @@ class State(RomAsset):
    __tablename__ = "states"
    __table_args__ = {"extend_existing": True}

-    emulator = Column(String(length=50), nullable=True)
+    emulator: Mapped[str | None] = mapped_column(String(length=50))

-    rom = relationship("Rom", lazy="joined", back_populates="states")
-    user = relationship("User", lazy="joined", back_populates="states")
+    rom: Mapped["Rom"] = relationship(lazy="joined", back_populates="states")
+    user: Mapped["User"] = relationship(lazy="joined", back_populates="states")

    @cached_property
    def screenshot(self) -> Optional["Screenshot"]:
@@ -93,5 +91,5 @@ class Screenshot(RomAsset):
    __tablename__ = "screenshots"
    __table_args__ = {"extend_existing": True}

-    rom = relationship("Rom", lazy="joined", back_populates="screenshots")
-    user = relationship("User", lazy="joined", back_populates="screenshots")
+    rom: Mapped["Rom"] = relationship(lazy="joined", back_populates="screenshots")
+    user: Mapped["User"] = relationship(lazy="joined", back_populates="screenshots")
--- a/backend/models/base.py
+++ b/backend/models/base.py
@@ -1,3 +1,4 @@
-from sqlalchemy.orm import declarative_base
+from sqlalchemy.orm import DeclarativeBase

-BaseModel = declarative_base()
+
+class BaseModel(DeclarativeBase): ...
--- a/backend/models/firmware.py
+++ b/backend/models/firmware.py
@@ -1,12 +1,16 @@
 import json
 import os
 from functools import cached_property
+from typing import TYPE_CHECKING

 from handler.metadata.base_hander import conditionally_set_cache
 from handler.redis_handler import cache
 from models.base import BaseModel
-from sqlalchemy import BigInteger, Column, ForeignKey, Integer, String
-from sqlalchemy.orm import relationship
+from sqlalchemy import BigInteger, ForeignKey, String
+from sqlalchemy.orm import Mapped, mapped_column, relationship
+
+if TYPE_CHECKING:
+    from models.platform import Platform

 KNOWN_BIOS_KEY = "romm:known_bios_files"
 conditionally_set_cache(
@@ -17,23 +21,25 @@ conditionally_set_cache(
 class Firmware(BaseModel):
    __tablename__ = "firmware"

-    id = Column(Integer(), primary_key=True, autoincrement=True)
-    platform_id = Column(
-        Integer(), ForeignKey("platforms.id", ondelete="CASCADE"), nullable=False
+    id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True)
+    platform_id: Mapped[int] = mapped_column(
+        ForeignKey("platforms.id", ondelete="CASCADE")
    )

-    file_name = Column(String(length=450), nullable=False)
-    file_name_no_tags = Column(String(length=450), nullable=False)
-    file_name_no_ext = Column(String(length=450), nullable=False)
-    file_extension = Column(String(length=100), nullable=False)
-    file_path = Column(String(length=1000), nullable=False)
-    file_size_bytes = Column(BigInteger(), default=0, nullable=False)
+    file_name: Mapped[str] = mapped_column(String(length=450))
+    file_name_no_tags: Mapped[str] = mapped_column(String(length=450))
+    file_name_no_ext: Mapped[str] = mapped_column(String(length=450))
+    file_extension: Mapped[str] = mapped_column(String(length=100))
+    file_path: Mapped[str] = mapped_column(String(length=1000))
+    file_size_bytes: Mapped[int] = mapped_column(BigInteger(), default=0)

-    crc_hash = Column(String(length=100), nullable=False)
-    md5_hash = Column(String(length=100), nullable=False)
-    sha1_hash = Column(String(length=100), nullable=False)
+    crc_hash: Mapped[str] = mapped_column(String(length=100))
+    md5_hash: Mapped[str] = mapped_column(String(length=100))
+    sha1_hash: Mapped[str] = mapped_column(String(length=100))

-    platform = relationship("Platform", lazy="joined", back_populates="firmware")
+    platform: Mapped["Platform"] = relationship(
+        lazy="joined", back_populates="firmware"
+    )

    @property
    def platform_slug(self) -> str:
--- a/backend/models/platform.py
+++ b/backend/models/platform.py
@@ -1,25 +1,29 @@
+from typing import TYPE_CHECKING
+
 from models.base import BaseModel
-from models.firmware import Firmware
 from models.rom import Rom
-from sqlalchemy import Column, Integer, String, func, select
-from sqlalchemy.orm import Mapped, column_property, relationship
+from sqlalchemy import String, func, select
+from sqlalchemy.orm import Mapped, column_property, mapped_column, relationship
+
+if TYPE_CHECKING:
+    from models.firmware import Firmware


 class Platform(BaseModel):
    __tablename__ = "platforms"

-    id = Column(Integer(), primary_key=True, autoincrement=True)
-    igdb_id: int = Column(Integer())
-    sgdb_id: int = Column(Integer())
-    moby_id: int = Column(Integer())
-    slug: str = Column(String(length=50), nullable=False)
-    fs_slug: str = Column(String(length=50), nullable=False)
-    name: str = Column(String(length=400))
-    logo_path: str = Column(String(length=1000), default="")
+    id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True)
+    igdb_id: Mapped[int | None]
+    sgdb_id: Mapped[int | None]
+    moby_id: Mapped[int | None]
+    slug: Mapped[str] = mapped_column(String(length=50))
+    fs_slug: Mapped[str] = mapped_column(String(length=50))
+    name: Mapped[str | None] = mapped_column(String(length=400))
+    logo_path: Mapped[str | None] = mapped_column(String(length=1000), default="")

-    roms: Mapped[set[Rom]] = relationship("Rom", back_populates="platform")
-    firmware: Mapped[set[Firmware]] = relationship(
-        "Firmware", lazy="selectin", back_populates="platform"
+    roms: Mapped[list["Rom"]] = relationship(back_populates="platform")
+    firmware: Mapped[list["Firmware"]] = relationship(
+        lazy="selectin", back_populates="platform"
    )

    # This runs a subquery to get the count of roms for the platform
--- a/backend/models/rom.py
+++ b/backend/models/rom.py
@@ -1,17 +1,14 @@
 from datetime import datetime
 from functools import cached_property
+from typing import TYPE_CHECKING, Any

 from config import FRONTEND_RESOURCES_PATH
-from models.assets import Save, Screenshot, State
 from models.base import BaseModel
 from sqlalchemy import (
    JSON,
    BigInteger,
-    Boolean,
-    Column,
    DateTime,
    ForeignKey,
-    Integer,
    String,
    Text,
    UniqueConstraint,
@@ -21,65 +18,69 @@ from sqlalchemy import (
    select,
 )
 from sqlalchemy.dialects.mysql.json import JSON as MySQLJSON
-from sqlalchemy.orm import Mapped, relationship
+from sqlalchemy.orm import Mapped, mapped_column, relationship
+
+if TYPE_CHECKING:
+    from models.assets import Save, Screenshot, State
+    from models.platform import Platform
+    from models.user import User


 class Rom(BaseModel):
    __tablename__ = "roms"

-    id = Column(Integer(), primary_key=True, autoincrement=True)
+    id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True)

-    igdb_id: int = Column(Integer())
-    sgdb_id: int = Column(Integer())
-    moby_id: int = Column(Integer())
+    igdb_id: Mapped[int | None]
+    sgdb_id: Mapped[int | None]
+    moby_id: Mapped[int | None]

-    file_name: str = Column(String(length=450), nullable=False)
-    file_name_no_tags: str = Column(String(length=450), nullable=False)
-    file_name_no_ext: str = Column(String(length=450), nullable=False)
-    file_extension: str = Column(String(length=100), nullable=False)
-    file_path: str = Column(String(length=1000), nullable=False)
-    file_size_bytes: int = Column(BigInteger(), default=0, nullable=False)
+    file_name: Mapped[str] = mapped_column(String(length=450))
+    file_name_no_tags: Mapped[str] = mapped_column(String(length=450))
+    file_name_no_ext: Mapped[str] = mapped_column(String(length=450))
+    file_extension: Mapped[str] = mapped_column(String(length=100))
+    file_path: Mapped[str] = mapped_column(String(length=1000))
+    file_size_bytes: Mapped[int] = mapped_column(BigInteger(), default=0)

-    name: str = Column(String(length=350))
-    slug: str = Column(String(length=400))
-    summary: str = Column(Text)
-    igdb_metadata: MySQLJSON = Column(MySQLJSON, default=dict)
-    moby_metadata: MySQLJSON = Column(MySQLJSON, default=dict)
+    name: Mapped[str | None] = mapped_column(String(length=350))
+    slug: Mapped[str | None] = mapped_column(String(length=400))
+    summary: Mapped[str | None] = mapped_column(Text)
+    igdb_metadata: Mapped[dict[str, Any] | None] = mapped_column(
+        MySQLJSON, default=dict
+    )
+    moby_metadata: Mapped[dict[str, Any] | None] = mapped_column(
+        MySQLJSON, default=dict
+    )

-    path_cover_s: str = Column(Text, default="")
-    path_cover_l: str = Column(Text, default="")
-    url_cover: str = Column(Text, default="", doc="URL to cover image stored in IGDB")
+    path_cover_s: Mapped[str | None] = mapped_column(Text, default="")
+    path_cover_l: Mapped[str | None] = mapped_column(Text, default="")
+    url_cover: Mapped[str | None] = mapped_column(
+        Text, default="", doc="URL to cover image stored in IGDB"
+    )

-    revision: str = Column(String(100))
-    regions: JSON = Column(JSON, default=[])
-    languages: JSON = Column(JSON, default=[])
-    tags: JSON = Column(JSON, default=[])
+    revision: Mapped[str | None] = mapped_column(String(100))
+    regions: Mapped[list[str] | None] = mapped_column(JSON, default=[])
+    languages: Mapped[list[str] | None] = mapped_column(JSON, default=[])
+    tags: Mapped[list[str] | None] = mapped_column(JSON, default=[])

-    path_screenshots: JSON = Column(JSON, default=[])
-    url_screenshots: JSON = Column(
+    path_screenshots: Mapped[list[str] | None] = mapped_column(JSON, default=[])
+    url_screenshots: Mapped[list[str] | None] = mapped_column(
        JSON, default=[], doc="URLs to screenshots stored in IGDB"
    )

-    multi: bool = Column(Boolean, default=False)
-    files: JSON = Column(JSON, default=[])
+    multi: Mapped[bool | None] = mapped_column(default=False)
+    files: Mapped[list[str] | None] = mapped_column(JSON, default=[])

-    platform_id = Column(
-        Integer(),
-        ForeignKey("platforms.id", ondelete="CASCADE"),
-        nullable=False,
+    platform_id: Mapped[int] = mapped_column(
+        ForeignKey("platforms.id", ondelete="CASCADE")
    )

-    platform = relationship("Platform", lazy="immediate")
+    platform: Mapped["Platform"] = relationship(lazy="immediate")

-    saves: Mapped[list[Save]] = relationship(
-        "Save",
-        back_populates="rom",
-    )
-    states: Mapped[list[State]] = relationship("State", back_populates="rom")
-    screenshots: Mapped[list[Screenshot]] = relationship(
-        "Screenshot", back_populates="rom"
-    )
-    notes: Mapped[list["RomNote"]] = relationship("RomNote", back_populates="rom")
+    saves: Mapped[list["Save"]] = relationship(back_populates="rom")
+    states: Mapped[list["State"]] = relationship(back_populates="rom")
+    screenshots: Mapped[list["Screenshot"]] = relationship(back_populates="rom")
+    notes: Mapped[list["RomNote"]] = relationship(back_populates="rom")

    @property
    def platform_slug(self) -> str:
@@ -184,26 +185,18 @@ class RomNote(BaseModel):
        UniqueConstraint("rom_id", "user_id", name="unique_rom_user_note"),
    )

-    id: int = Column(Integer(), primary_key=True, autoincrement=True)
-    last_edited_at: datetime = Column(
-        DateTime(timezone=True), server_default=func.now(), nullable=False
+    id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True)
+    last_edited_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), server_default=func.now()
    )
-    raw_markdown: str = Column(Text, nullable=False, default="")
-    is_public: bool = Column(Boolean, default=False)
+    raw_markdown: Mapped[str] = mapped_column(Text, default="")
+    is_public: Mapped[bool | None] = mapped_column(default=False)

-    rom_id: int = Column(
-        Integer(),
-        ForeignKey("roms.id", ondelete="CASCADE"),
-        nullable=False,
-    )
-    user_id: int = Column(
-        Integer(),
-        ForeignKey("users.id", ondelete="CASCADE"),
-        nullable=False,
-    )
+    rom_id: Mapped[int] = mapped_column(ForeignKey("roms.id", ondelete="CASCADE"))
+    user_id: Mapped[int] = mapped_column(ForeignKey("users.id", ondelete="CASCADE"))

-    rom = relationship("Rom", lazy="joined", back_populates="notes")
-    user = relationship("User", lazy="joined", back_populates="notes")
+    rom: Mapped["Rom"] = relationship(lazy="joined", back_populates="notes")
+    user: Mapped["User"] = relationship(lazy="joined", back_populates="notes")

    @property
    def user__username(self) -> str:
--- a/backend/models/user.py
+++ b/backend/models/user.py
@@ -1,12 +1,16 @@
-import datetime
 import enum
+from datetime import datetime
+from typing import TYPE_CHECKING

-from models.assets import Save, Screenshot, State
 from models.base import BaseModel
-from sqlalchemy import Boolean, Column, DateTime, Enum, Integer, String
-from sqlalchemy.orm import Mapped, relationship
+from sqlalchemy import DateTime, Enum, String
+from sqlalchemy.orm import Mapped, mapped_column, relationship
 from starlette.authentication import SimpleUser

+if TYPE_CHECKING:
+    from models.assets import Save, Screenshot, State
+    from models.rom import RomNote
+

 class Role(enum.Enum):
    VIEWER = "viewer"
@@ -15,30 +19,25 @@ class Role(enum.Enum):


 class User(BaseModel, SimpleUser):
-    from models.rom import RomNote
-
    __tablename__ = "users"
    __table_args__ = {"extend_existing": True}

-    id = Column(Integer(), primary_key=True, autoincrement=True)
+    id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True)

-    username: str = Column(String(length=255), unique=True, index=True)
-    hashed_password: str = Column(String(length=255))
-    enabled: bool = Column(Boolean(), default=True)
-    role: Role = Column(Enum(Role), default=Role.VIEWER)
-    avatar_path: str = Column(String(length=255), default="")
-    last_login: Mapped[datetime] = Column(DateTime(timezone=True), nullable=True)
-    last_active: Mapped[datetime] = Column(DateTime(timezone=True), nullable=True)
+    username: Mapped[str | None] = mapped_column(
+        String(length=255), unique=True, index=True
+    )
+    hashed_password: Mapped[str | None] = mapped_column(String(length=255))
+    enabled: Mapped[bool | None] = mapped_column(default=True)
+    role: Mapped[Role | None] = mapped_column(Enum(Role), default=Role.VIEWER)
+    avatar_path: Mapped[str | None] = mapped_column(String(length=255), default="")
+    last_login: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
+    last_active: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))

-    saves: Mapped[list[Save]] = relationship(
-        "Save",
-        back_populates="user",
-    )
-    states: Mapped[list[State]] = relationship("State", back_populates="user")
-    screenshots: Mapped[list[Screenshot]] = relationship(
-        "Screenshot", back_populates="user"
-    )
-    notes: Mapped[list[RomNote]] = relationship("RomNote", back_populates="user")
+    saves: Mapped[list["Save"]] = relationship(back_populates="user")
+    states: Mapped[list["State"]] = relationship(back_populates="user")
+    screenshots: Mapped[list["Screenshot"]] = relationship(back_populates="user")
+    notes: Mapped[list["RomNote"]] = relationship(back_populates="user")

    @property
    def oauth_scopes(self):
@@ -60,4 +59,4 @@ class User(BaseModel, SimpleUser):
    def set_last_active(self):
        from handler.database import db_user_handler

-        db_user_handler.update_user(self.id, {"last_active": datetime.datetime.now()})
+        db_user_handler.update_user(self.id, {"last_active": datetime.now()})
--- a/poetry.lock
+++ b/poetry.lock
@@ -1856,7 +1856,6 @@ files = [

 [package.dependencies]
 greenlet = {version = "!=0.4.17", markers = "platform_machine == \"aarch64\" or platform_machine == \"ppc64le\" or platform_machine == \"x86_64\" or platform_machine == \"amd64\" or platform_machine == \"AMD64\" or platform_machine == \"win32\" or platform_machine == \"WIN32\""}
-mypy = {version = ">=0.910", optional = true, markers = "extra == \"mypy\""}
 typing-extensions = ">=4.6.0"

 [package.extras]
@@ -2467,4 +2466,4 @@ multidict = ">=4.0"
 [metadata]
 lock-version = "2.0"
 python-versions = "^3.11"
-content-hash = "e6bb882d0b044da78ee8fe6415d70228ee4d00511c7e3b89d64c3b0a5ae0f0bf"
+content-hash = "05030ed3494abd524269977863f2508bbee9324cfd01253a96684a87fdc5add6"
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -15,7 +15,7 @@ uvicorn = "0.29.0"
 gunicorn = "22.0.0"
 websockets = "12.0"
 python-socketio = "5.11.1"
-SQLAlchemy = { extras = ["mypy"], version = "^2.0.30" }
+SQLAlchemy = "^2.0.30"
 alembic = "1.13.1"
 PyYAML = "6.0.1"
 Unidecode = "1.3.8"