Models and API refactor

backend/alembic/env.py

@@ -47,6 +47,7 @@ def run_migrations_offline() -> None:
         render_as_batch=True,
         literal_binds=True,
         dialect_opts={"paramstyle": "named"},
+        compare_type=True,
     )
 
     with context.begin_transaction():
@@ -65,7 +66,10 @@ def run_migrations_online() -> None:
 
     with engine.connect() as connection:
         context.configure(
-            connection=connection, target_metadata=target_metadata, render_as_batch=True
+            connection=connection,
+            target_metadata=target_metadata,
+            render_as_batch=True,
+            compare_type=True,
         )
 
         with context.begin_transaction():

backend/alembic/versions/0009_models_refactor.py

@@ -0,0 +1,156 @@
+"""empty message
+
+Revision ID: 0009_models_refactor
+Revises: 1.9.2
+Create Date: 2023-09-12 18:18:27.158732
+
+"""
+from alembic import op
+import sqlalchemy as sa
+from sqlalchemy.dialects import mysql
+
+# revision identifiers, used by Alembic.
+revision = "0009_models_refactor"
+down_revision = "1.9.2"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    with op.batch_alter_table("platforms", schema=None) as batch_op:
+        batch_op.alter_column(
+            "igdb_id", existing_type=mysql.VARCHAR(length=10), nullable=True
+        )
+        batch_op.alter_column(
+            "sgdb_id", existing_type=mysql.VARCHAR(length=10), nullable=True
+        )
+        batch_op.alter_column(
+            "slug", existing_type=mysql.VARCHAR(length=50), nullable=False
+        )
+        batch_op.alter_column(
+            "name", existing_type=mysql.VARCHAR(length=400), nullable=True
+        )
+
+        # Move primary key to slug
+        batch_op.drop_constraint("PRIMARY", type_="primary")
+        batch_op.create_primary_key(None, ["slug"])
+
+    with op.batch_alter_table("roms", schema=None) as batch_op:
+        batch_op.alter_column(
+            "r_igdb_id",
+            existing_type=mysql.VARCHAR(length=10),
+            new_column_name="igdb_id",
+        )
+        batch_op.alter_column(
+            "r_sgdb_id",
+            existing_type=mysql.VARCHAR(length=10),
+            new_column_name="sgdb_id",
+        )
+        batch_op.alter_column(
+            "r_slug", existing_type=mysql.VARCHAR(length=400), new_column_name="slug"
+        )
+        batch_op.alter_column(
+            "r_name", existing_type=mysql.VARCHAR(length=350), new_column_name="name"
+        )
+        batch_op.alter_column(
+            "p_slug",
+            existing_type=mysql.VARCHAR(length=50),
+            new_column_name="platform_slug",
+            nullable=False,
+        )
+
+        batch_op.alter_column(
+            "file_extension", existing_type=mysql.VARCHAR(length=10), nullable=False
+        )
+        batch_op.alter_column(
+            "file_path", existing_type=mysql.VARCHAR(length=1000), nullable=False
+        )
+        batch_op.alter_column("file_size", existing_type=mysql.FLOAT(), nullable=False)
+        batch_op.alter_column(
+            "file_size_units", existing_type=mysql.VARCHAR(length=10), nullable=False
+        )
+        batch_op.alter_column(
+            "url_screenshots",
+            existing_type=mysql.LONGTEXT(charset="utf8mb4", collation="utf8mb4_bin"),
+            nullable=True,
+            existing_server_default=sa.text("'[]'"),
+        )
+        batch_op.alter_column(
+            "path_screenshots",
+            existing_type=mysql.LONGTEXT(charset="utf8mb4", collation="utf8mb4_bin"),
+            nullable=True,
+            existing_server_default=sa.text("'[]'"),
+        )
+
+        batch_op.create_foreign_key(
+            "fk_platform_roms", "platforms", ["platform_slug"], ["slug"]
+        )
+
+
+def downgrade() -> None:
+    with op.batch_alter_table("roms", schema=None) as batch_op:
+        batch_op.alter_column(
+            "igdb_id",
+            existing_type=mysql.VARCHAR(length=10),
+            new_column_name="r_igdb_id",
+        )
+        batch_op.alter_column(
+            "sgdb_id",
+            existing_type=mysql.VARCHAR(length=10),
+            new_column_name="r_sgdb_id",
+        )
+        batch_op.alter_column(
+            "slug", existing_type=mysql.VARCHAR(length=400), new_column_name="r_slug"
+        )
+        batch_op.alter_column(
+            "name", existing_type=mysql.VARCHAR(length=350), new_column_name="r_name"
+        )
+        batch_op.alter_column(
+            "platform_slug",
+            existing_type=mysql.VARCHAR(length=50),
+            new_column_name="p_slug",
+            nullable=True,
+        )
+
+        batch_op.alter_column(
+            "path_screenshots",
+            existing_type=mysql.LONGTEXT(charset="utf8mb4", collation="utf8mb4_bin"),
+            nullable=False,
+            existing_server_default=sa.text("'[]'"),
+        )
+        batch_op.alter_column(
+            "url_screenshots",
+            existing_type=mysql.LONGTEXT(charset="utf8mb4", collation="utf8mb4_bin"),
+            nullable=False,
+            existing_server_default=sa.text("'[]'"),
+        )
+        batch_op.alter_column(
+            "file_size_units", existing_type=mysql.VARCHAR(length=10), nullable=True
+        )
+        batch_op.alter_column("file_size", existing_type=mysql.FLOAT(), nullable=True)
+        batch_op.alter_column(
+            "file_path", existing_type=mysql.VARCHAR(length=1000), nullable=True
+        )
+        batch_op.alter_column(
+            "file_extension", existing_type=mysql.VARCHAR(length=10), nullable=True
+        )
+
+        batch_op.drop_constraint("fk_platform_roms", type_="foreignkey")
+
+    with op.batch_alter_table("platforms", schema=None) as batch_op:
+        batch_op.alter_column(
+            "igdb_id", existing_type=mysql.VARCHAR(length=10), nullable=False
+        )
+        batch_op.alter_column(
+            "sgdb_id", existing_type=mysql.VARCHAR(length=10), nullable=False
+        )
+        batch_op.alter_column(
+            "slug", existing_type=mysql.VARCHAR(length=50), nullable=True
+        )
+        batch_op.alter_column(
+            "name", existing_type=mysql.VARCHAR(length=400), nullable=False
+        )
+
+        # Move primary key to slug
+        batch_op.drop_constraint("PRIMARY", type_="primary")
+        batch_op.create_primary_key(None, ["fs_slug"])

backend/alembic/versions/0010_igdb_id_integerr.py

@@ -0,0 +1,88 @@
+"""empty message
+
+Revision ID: 0010_igdb_id_integerr
+Revises: 0009_models_refactor
+Create Date: 2023-09-14 09:57:13.487331
+
+"""
+from alembic import op
+import sqlalchemy as sa
+from sqlalchemy.dialects import mysql
+
+# revision identifiers, used by Alembic.
+revision = "0010_igdb_id_integerr"
+down_revision = "0009_models_refactor"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    # ### commands auto generated by Alembic - please adjust! ###
+    with op.batch_alter_table("platforms", schema=None) as batch_op:
+        batch_op.execute('UPDATE platforms SET igdb_id = NULL WHERE igdb_id = ""')
+        batch_op.execute('UPDATE platforms SET sgdb_id = NULL WHERE sgdb_id = ""')
+
+        batch_op.alter_column(
+            "igdb_id",
+            existing_type=mysql.VARCHAR(length=10),
+            type_=sa.Integer(),
+            existing_nullable=True,
+        )
+        batch_op.alter_column(
+            "sgdb_id",
+            existing_type=mysql.VARCHAR(length=10),
+            type_=sa.Integer(),
+            existing_nullable=True,
+        )
+
+    with op.batch_alter_table("roms", schema=None) as batch_op:
+        batch_op.execute('UPDATE roms SET igdb_id = NULL WHERE igdb_id = ""')
+        batch_op.execute('UPDATE roms SET sgdb_id = NULL WHERE sgdb_id = ""')
+
+        batch_op.alter_column(
+            "igdb_id",
+            existing_type=mysql.VARCHAR(length=10),
+            type_=sa.Integer(),
+            existing_nullable=True,
+        )
+        batch_op.alter_column(
+            "sgdb_id",
+            existing_type=mysql.VARCHAR(length=10),
+            type_=sa.Integer(),
+            existing_nullable=True,
+        )
+
+    # ### end Alembic commands ###
+
+
+def downgrade() -> None:
+    # ### commands auto generated by Alembic - please adjust! ###
+    with op.batch_alter_table("roms", schema=None) as batch_op:
+        batch_op.alter_column(
+            "sgdb_id",
+            existing_type=sa.Integer(),
+            type_=mysql.VARCHAR(length=10),
+            existing_nullable=True,
+        )
+        batch_op.alter_column(
+            "igdb_id",
+            existing_type=sa.Integer(),
+            type_=mysql.VARCHAR(length=10),
+            existing_nullable=True,
+        )
+
+    with op.batch_alter_table("platforms", schema=None) as batch_op:
+        batch_op.alter_column(
+            "sgdb_id",
+            existing_type=sa.Integer(),
+            type_=mysql.VARCHAR(length=10),
+            existing_nullable=True,
+        )
+        batch_op.alter_column(
+            "igdb_id",
+            existing_type=sa.Integer(),
+            type_=mysql.VARCHAR(length=10),
+            existing_nullable=True,
+        )
+
+    # ### end Alembic commands ###

backend/endpoints/identity.py

@@ -1,8 +1,9 @@
 import secrets
 from typing import Optional, Annotated
+from typing_extensions import TypedDict
 from fastapi import APIRouter, HTTPException, status, Request, Depends, File, UploadFile
 from fastapi.security.http import HTTPBasic
-from pydantic import BaseModel, BaseConfig
+from pydantic import BaseModel, ConfigDict
 
 from handler import dbh
 from models.user import User, Role
@@ -17,19 +18,22 @@
 
 
 class UserSchema(BaseModel):
+    model_config = ConfigDict(from_attributes=True)
+
     id: int
     username: str
     enabled: bool
     role: Role
     oauth_scopes: list[str]
     avatar_path: str
 
-    class Config(BaseConfig):
-        orm_mode = True
+
+class MessageResponse(TypedDict):
+    message: str
 
 
 @router.post("/login")
-def login(request: Request, credentials=Depends(HTTPBasic())):
+def login(request: Request, credentials=Depends(HTTPBasic())) -> MessageResponse:
     """Session login endpoint"""
     user = authenticate_user(credentials.username, credentials.password)
     if not user:
@@ -46,7 +50,7 @@ def login(request: Request, credentials=Depends(HTTPBasic())):
 
 
 @router.post("/logout")
-def logout(request: Request):
+def logout(request: Request) -> MessageResponse:
     """Session logout endpoint"""
     # Check if session key already stored in cache
     session_id = request.session.get("session_id")
@@ -182,7 +186,7 @@ def update_user(
 
 
 @protected_route(router.delete, "/users/{user_id}", ["users.write"])
-def delete_user(request: Request, user_id: int):
+def delete_user(request: Request, user_id: int) -> MessageResponse:
     """Delete a specific user"""
     if not ROMM_AUTH_ENABLED:
         raise HTTPException(

backend/endpoints/oauth.py

@@ -1,4 +1,5 @@
 from typing import Annotated, Final
+from typing_extensions import TypedDict, NotRequired
 from datetime import timedelta
 from fastapi import Depends, APIRouter, HTTPException, status
 
@@ -17,8 +18,15 @@
 router = APIRouter()
 
 
+class TokenResponse(TypedDict):
+    access_token: str
+    refresh_token: NotRequired[str]
+    token_type: str
+    expires: int
+
+
 @router.post("/token")
-async def token(form_data: Annotated[OAuth2RequestForm, Depends()]):
+async def token(form_data: Annotated[OAuth2RequestForm, Depends()]) -> TokenResponse:
     """OAuth2 token endpoint"""
 
     # Suppport refreshing access tokens

backend/endpoints/platform.py

@@ -1,5 +1,6 @@
 from fastapi import APIRouter, Request
-from pydantic import BaseModel, BaseConfig
+from pydantic import BaseModel, ConfigDict
+from typing import Optional
 
 from handler import dbh
 from utils.oauth import protected_route
@@ -8,19 +9,15 @@
 
 
 class PlatformSchema(BaseModel):
-    igdb_id: str
-    sgdb_id: str
+    model_config = ConfigDict(from_attributes=True)
 
     slug: str
-    name: str
-
-    logo_path: str
     fs_slug: str
-
-    n_roms: int
-
-    class Config(BaseConfig):
-        orm_mode = True
+    igdb_id: Optional[int] = None
+    sgdb_id: Optional[int] = None
+    name: Optional[str]
+    logo_path: str
+    rom_count: int
 
 
 @protected_route(router.get, "/platforms", ["platforms.read"])