Push notifications for new tags and imported folders

backend/beets_flask/config/beets_config.py

@@ -26,6 +26,7 @@
 """
 
 import os
+from pathlib import Path
 
 from beets import IncludeLazyConfig as BeetsConfig
 from beets.plugins import load_plugins
@@ -89,9 +90,7 @@ def reset(self):
 
         # Load config from default location (set via env var)
         # if it is set otherwise use the default location
-        ib_folder = os.getenv("BEETSFLASKDIR")
-        if ib_folder is None:
-            ib_folder = os.path.expanduser("~/.config/beets-flask")
+        ib_folder = get_bf_config_dir()
         ib_config_path = os.path.join(ib_folder, "config.yaml")
 
         # Check if the user config exists
@@ -193,6 +192,16 @@ def get_config(force_refresh=False) -> InteractiveBeetsConfig:
     return config
 
 
-__all__ = ["refresh_config", "get_config"]
+def get_bf_config_dir() -> Path:
+    """Get the path to the beets-flask config directory."""
+    # This is the directory where the config.yaml file is stored
+    # and where the vapid keys are stored.
+
+    return Path(
+        os.getenv("BEETSFLASKDIR", os.path.expanduser("~/.config/beets-flask"))
+    ).resolve()
+
+
+__all__ = ["refresh_config", "get_config", "get_bf_config_dir"]
 
 # raise NotImplementedError("This module should not be imported.")

backend/beets_flask/database/models/__init__.py

@@ -1,4 +1,5 @@
 from .base import Base
+from .notifications import PushSubscription, SubscriptionSettings, WebhookSubscription
 from .states import CandidateStateInDb, FolderInDb, SessionStateInDb, TaskStateInDb
 
 __all__ = [
@@ -7,4 +8,7 @@
     "SessionStateInDb",
     "TaskStateInDb",
     "CandidateStateInDb",
+    "PushSubscription",
+    "SubscriptionSettings",
+    "WebhookSubscription",
 ]

backend/beets_flask/database/models/notifications.py

@@ -0,0 +1,144 @@
+from __future__ import annotations
+
+from enum import Enum
+from typing import Any, Literal, Mapping
+
+from sqlalchemy import ForeignKey
+from sqlalchemy.orm import Mapped, mapped_column, relationship
+
+from .base import Base
+
+
+class SubscriptionSettings(Base):
+    """Represents options for a push subscription.
+
+    This class is used to store additional options for a push subscription,
+    such as when to trigger this notification.
+    """
+
+    __tablename__ = "push_settings"
+
+    is_active: Mapped[bool]
+
+    def __init__(self, is_active: bool = True):
+        super().__init__()
+        self.is_active = is_active
+
+    def update_from_dict(self, data: dict[str, Any]):
+        """Update the PushSettings instance from a dictionary.
+
+        None values are ignored.
+        """
+        is_active = data.get("is_active")
+        self.is_active = is_active if is_active is not None else self.is_active
+
+    @classmethod
+    def from_dict(cls, data: dict[str, Any]) -> SubscriptionSettings:
+        """Create a PushSettings instance from a dictionary."""
+        instance = cls()
+        instance.update_from_dict(data)
+        return instance
+
+
+class PushSubscription(Base):
+    """Represents a push subscription.
+
+    Expects push subscriptions in Web Push API,
+    https://developer.mozilla.org/en-US/docs/Web/API/Push_API.
+    """
+
+    __tablename__ = "push_subscription"
+
+    # id==endpoint
+    keys: Mapped[dict[str, str]]
+    expiration_time: Mapped[int | None]
+
+    # Settings on when to trigger this
+    settings_id: Mapped[str] = mapped_column(ForeignKey("push_settings.id"), index=True)
+    settings: Mapped[SubscriptionSettings] = relationship(
+        foreign_keys=[settings_id], cascade="all"
+    )
+
+    def __init__(
+        self,
+        id: str,
+        keys: dict[str, str] | None = None,
+        expiration_time: int | None = None,
+        settings: SubscriptionSettings | None = None,
+    ):
+        super().__init__(id=id)
+        self.keys = keys or {}
+        self.expiration_time = expiration_time
+        self.settings = settings or SubscriptionSettings()
+
+    @property
+    def endpoint(self) -> str:
+        """
+        Convenience property to get the id.
+
+        Note: Although the id is just the endpoint, when querying the db, you **must** use `PushSubscription.id == endpoint`. Sqlalchemy does not resolve properties.
+        """
+        return self.id
+
+    def to_dict(self) -> Mapping:
+        col_map = super().to_dict()
+        return {
+            **col_map,
+            "settings": self.settings.to_dict(),
+        }
+
+
+class WebhookType(Enum):
+    WEBPUSH = 0
+
+
+class WebhookSubscription(Base):
+    """Webhook handlers for push notifications.
+
+    Additionally to :class:`PushSubscription`, this class can be used to handle push notifications
+    to generic endpoints, such as a webhook URL.
+    """
+
+    __tablename__ = "push_webhooks"
+
+    type: Mapped[WebhookType]
+    # Required fields for push webhooks
+    url: Mapped[str]
+    method: Mapped[str]  # e.g., "POST", "GET"
+
+    # Optional fields for push webhooks
+    headers: Mapped[dict[str, str] | None]
+    params: Mapped[dict[str, str] | None]
+    body: Mapped[dict[str, Any] | None]
+
+    # Settings on when to trigger this
+    settings_id: Mapped[str] = mapped_column(ForeignKey("push_settings.id"), index=True)
+    settings: Mapped[SubscriptionSettings] = relationship(
+        foreign_keys=[settings_id], cascade="all"
+    )
+
+    def __init__(
+        self,
+        url: str,
+        method: str = "POST",
+        headers: dict[str, str] | None = None,
+        params: dict[str, str] | None = None,
+        body: dict[str, Any] | None = None,
+        settings: SubscriptionSettings | None = None,
+    ):
+        """Initialize a PushWebHooks instance."""
+        super().__init__()
+        self.type = WebhookType.WEBPUSH
+        self.url = url
+        self.method = method
+        self.headers = headers
+        self.params = params
+        self.body = body
+        self.settings = settings or SubscriptionSettings()
+
+    def to_dict(self) -> Mapping:
+        col_map = super().to_dict()
+        return {
+            **col_map,
+            "settings": self.settings.to_dict(),
+        }

backend/beets_flask/server/routes/__init__.py

@@ -8,6 +8,7 @@
 from .inbox import inbox_bp
 from .library import library_bp
 from .monitor import monitor_bp
+from .notifications import register_notifications
 
 backend_bp = Blueprint("backend", __name__, url_prefix="/api_v1")
 
@@ -26,6 +27,9 @@ def register_routes(app: Quart):
     # to api blueprint i.e. /api_v1/session, /api_v1/task & /api_v1/candidate
     register_state_models(backend_bp)
 
+    # notifications/
+    register_notifications(backend_bp)
+
     app.register_blueprint(backend_bp)
     app.register_blueprint(frontend_bp)
 

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

@@ -41,8 +41,8 @@ def __init__(self, model: type[T], url_prefix: str | None = None):
     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)
+        self.blueprint.route("/id/<path:id>", methods=["GET"])(self.get_by_id)
+        self.blueprint.route("/id/<path:id>", methods=["DELETE"])(self.delete_by_id)
 
     async def get_all(self):
         params = dict(request.args)