Added sqlalchemy types to allow inserting generic dicts as json

CHANGELOG.md

@@ -12,6 +12,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - Configuration option for artist separator characters `gui.library.artist_separator`
 - Docs subpage for configuration (including content)
 - `typing_extensions` is now a dependency, to allow for more typing features
+- The model api routes now allows for `DELETE` requests to delete resources by id. Not used yet but will be helpful for future features.
 
 ### Fixed
 
@@ -23,6 +24,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - In albums and items view the clicking on artists does not return any results if the contained a separator character (e.g. `&`) [#132](https://github.com/pSpitzner/beets-flask/issues/138)
 - Cleanup old actions.tsx file, which included old unused code [#134](https://github.com/pSpitzner/beets-flask/issues/134)
 
+### Changed
+
+- Created `types.py` file to hold custom sqlalchemy types, and moved `IntDictType` there.
+
 ## [1.0.0] - 25-07-06
 
 This is a breaking change, you will need to update your configs and delete your beets-flask

backend/beets_flask/database/models/base.py

@@ -1,12 +1,11 @@
 from __future__ import annotations
 
-import json
 from datetime import datetime
-from typing import Mapping, Self, Sequence
+from typing import Any, Mapping, Self, Sequence
 from uuid import uuid4
 
 import pytz
-from sqlalchemy import LargeBinary, select, types
+from sqlalchemy import LargeBinary, select
 from sqlalchemy.orm import (
     DeclarativeBase,
     Mapped,
@@ -19,36 +18,19 @@
 
 from beets_flask.logger import log
 
-
-class IntDictType(types.TypeDecorator):
-    """Stores a dict[int, int] as a JSON-encoded string in the database."""
-
-    impl = types.Text
-    cache_ok = True
-
-    def process_bind_param(self, value, dialect):
-        if value is None:
-            return None
-        if not isinstance(value, dict) or not all(
-            isinstance(k, int) and isinstance(v, int) for k, v in value.items()
-        ):
-            raise ValueError("Value must be a dict[int, int]")
-        return json.dumps({str(k): v for k, v in value.items()})
-
-    def process_result_value(self, value, dialect):
-        if value is None:
-            return None
-        return {int(k): v for k, v in json.loads(value).items()}
-
-    def copy(self, **kw):
-        return IntDictType(self.impl.length)  # type: ignore
+from .types import DictType, IntDictType, StrDictType
 
 
 class Base(DeclarativeBase):
     __abstract__ = True
 
     registry = registry(
-        type_annotation_map={bytes: LargeBinary, dict[int, int]: IntDictType}
+        type_annotation_map={
+            bytes: LargeBinary,
+            dict[int, int]: IntDictType,
+            dict[str, str]: StrDictType,
+            dict[str, Any]: DictType,
+        }
     )
 
     id: Mapped[str] = mapped_column(primary_key=True)

backend/beets_flask/database/models/types.py

@@ -0,0 +1,67 @@
+import json
+from typing import Any
+
+from sqlalchemy import types
+
+
+class DictType(types.TypeDecorator):
+    """Stores a dict[str, Any] as a JSON-encoded string in the database.
+
+    Allows for flexible storage of dictionaries with string keys and values of
+    any (serializable) type.
+    """
+
+    impl = types.Text
+    cache_ok = True
+
+    allowed_keys_types: tuple[type, ...] = (str,)
+    allowed_values_types: tuple[type | Any, ...] = (Any,)
+
+    def process_bind_param(self, value, dialect):
+        if value is None:
+            return None
+        if not isinstance(value, dict):
+            raise ValueError("Value must be a dict")
+
+        # Any type needs some special handling
+        allowed_types_v: tuple[type, ...] = tuple(
+            filter(lambda x: x is not Any, self.allowed_values_types)
+        )
+
+        if not len(allowed_types_v) == 0:
+            if not all(isinstance(v, allowed_types_v) for v in value.values()):
+                raise ValueError(
+                    f"Value must be a dict with values of type {allowed_types_v}. Got: {value.values()}"
+                )
+
+        if not all(isinstance(k, self.allowed_keys_types) for k in value.keys()):
+            raise ValueError(f"Keys must be of type {self.allowed_keys_types}.")
+
+        return json.dumps({str(k): v for k, v in value.items()})
+
+    def process_result_value(self, value, dialect):
+        if value is None:
+            return None
+        return json.loads(value)
+
+    def copy(self, **kw):
+        return self.__class__(self.impl.length)  # type: ignore
+
+
+class IntDictType(DictType):
+    """Stores a dict[int, int] as a JSON-encoded string in the database."""
+
+    allowed_keys_types = (int,)
+    allowed_values_types = (int,)
+
+    def process_result_value(self, value, dialect):
+        if value is None:
+            return None
+        return {int(k): int(v) for k, v in json.loads(value).items()}
+
+
+class StrDictType(DictType):
+    """Stores a dict[str, str] as a JSON-encoded string in the database."""
+
+    allowed_keys_types = (str,)
+    allowed_values_types = (str,)

backend/beets_flask/server/routes/db_models/base.py

@@ -1,5 +1,5 @@
 from datetime import datetime
-from typing import Any, Generic, Sequence, TypeVar
+from typing import Generic, Sequence, TypeVar
 
 from quart import Blueprint, request
 from sqlalchemy import select
@@ -42,6 +42,7 @@ def _register_routes(self) -> None:
         """Register the routes for the blueprint."""
         self.blueprint.route("/", methods=["GET"])(self.get_all)
         self.blueprint.route("/id/<id>", methods=["GET"])(self.get_by_id)
+        self.blueprint.route("/id/<id>", methods=["DELETE"])(self.delete_by_id)
 
     async def get_all(self):
         params = dict(request.args)
@@ -79,6 +80,16 @@ async def get_by_id(self, id: str):
 
             return item.to_dict()
 
+    async def delete_by_id(self, id: str):
+        with db_session_factory() as session:
+            item = self.model.get_by(self.model.id == id, session=session)
+            if not item:
+                return {"message": f"Item with id {id} not found"}, 200
+            session.delete(item)
+            session.commit()
+
+        return {"message": f"Item with id {id} deleted successfully"}, 200
+
 
 # ------------------------------- Local Utility ------------------------------ #