Use Pydantic for settings validation data model (#204)

* Use Pydantic data models for settings

* Rename Scraping to Scraper to be consistent across platform

* Revert "Rename Scraping to Scraper to be consistent across platform"

This reverts commit db29f42.

* fix data model

* Fix data model and references

* Add url field to torrentio model

* Correct docstring

Author: omnunum

backend/controllers/default.py

@@ -1,6 +1,6 @@
 from fastapi import APIRouter, Request
 import requests
-from utils.settings import settings_manager
+from program.settings.manager import settings_manager
 
 
 router = APIRouter(
@@ -26,7 +26,7 @@ async def health(request: Request):
 
 @router.get("/user")
 async def get_rd_user():
-    api_key = settings_manager.get("real_debrid.api_key")
+    api_key = settings_manager.settings.real_debrid.api_key
     headers = {"Authorization": f"Bearer {api_key}"}
     response = requests.get(
         "https://api.real-debrid.com/rest/1.0/user", headers=headers

backend/controllers/settings.py

@@ -1,6 +1,6 @@
 from copy import copy
-from fastapi import APIRouter
-from utils.settings import settings_manager
+from fastapi import APIRouter, HTTPException
+from program.settings.manager import settings_manager
 from pydantic import BaseModel
 from typing import Any, List
 
@@ -39,14 +39,25 @@ async def save_settings():
 async def get_all_settings():
     return {
         "success": True,
-        "data": copy(settings_manager.get_all()),
+        "data": copy(settings_manager.settings),
     }
 
 
-@router.get("/get/{keys}")
-async def get_settings(keys: str):
-    keys = keys.split(",")
-    data = {key: settings_manager.get(key) for key in keys}
+@router.get("/get/{paths}")
+async def get_settings(paths: str):
+    current_settings = settings_manager.settings.dict()
+    data = {}
+    for path in paths.split(","):
+        keys = path.split('.')
+        current_obj = current_settings
+
+        for k in keys:
+            if k not in current_obj:
+                return None
+            current_obj = current_obj[k]
+        
+        data[path] = current_obj
+
     return {
         "success": True,
         "data": data,
@@ -55,8 +66,33 @@ async def get_settings(keys: str):
 
 @router.post("/set")
 async def set_settings(settings: List[SetSettings]):
-    settings_manager.set(settings)
+    current_settings = settings_manager.settings.dict()
+    
+    for setting in settings:
+        keys = setting.key.split('.')
+        current_obj = current_settings
+        
+        # Navigate to the last key's parent object, similar to the getter.
+        for k in keys[:-1]:
+            if k not in current_obj:
+                # If a key in the path does not exist, raise an exception or optionally create a new dict.
+                raise HTTPException(status_code=400, detail=f"Path '{'.'.join(keys[:-1])}' does not exist.")
+            current_obj = current_obj[k]
+        
+        # Set the value at the final key.
+        if keys[-1] in current_obj:
+            current_obj[keys[-1]] = setting.value
+        else:
+            # If the final key does not exist, raise an exception.
+            raise HTTPException(status_code=400, detail=f"Key '{keys[-1]}' does not exist in path '{'.'.join(keys[:-1])}'.")
+    
+
+    settings_manager.load(settings_dict=current_settings)
+    
+    # Notify observers about the update.
+    settings_manager.notify_observers()
+    
     return {
         "success": True,
-        "message": "Settings saved!",
-    }
+        "message": "Settings updated successfully."
+    }

backend/program/__init__.py

@@ -7,11 +7,12 @@
 from program.realdebrid import Debrid
 from program.symlink import Symlinker
 from program.media.container import MediaItemContainer
-from utils.logger import logger, get_data_path
+from utils.logger import logger
 from program.plex import Plex
 from program.content import Content
 from utils.utils import Pickly
-from utils.settings import settings_manager as settings
+from utils import data_dir_path
+from program.settings.manager import settings_manager
 from utils.service_manager import ServiceManager
 
 
@@ -22,14 +23,17 @@ def __init__(self, args):
         super().__init__(name="Iceberg")
         self.running = False
         self.startup_args = args
+        logger.configure_logger(
+            debug=settings_manager.settings.debug,
+            log=settings_manager.settings.log
+        )
 
     def start(self):
-        logger.info("Iceberg v%s starting!", settings.get("version"))
+        logger.info("Iceberg v%s starting!", settings_manager.settings.version)
         self.initialized = False
         self.media_items = MediaItemContainer(items=[])
-        self.data_path = get_data_path()
-        if not os.path.exists(self.data_path):
-            os.mkdir(self.data_path)
+        self.data_path = data_dir_path
+        os.makedirs(self.data_path, exist_ok=True)
         if not self.startup_args.dev:
             self.pickly = Pickly(self.media_items, self.data_path)
             self.pickly.start()

backend/program/content/listrr.py

@@ -1,30 +1,23 @@
-"""Mdblist content module"""
+"""Listrr content module"""
 from time import time
 from typing import Optional
-from pydantic import BaseModel
-from utils.settings import settings_manager
+
+from requests.exceptions import HTTPError
+
+from program.settings.manager import settings_manager
 from utils.logger import logger
 from utils.request import get, ping
-from requests.exceptions import HTTPError
 from program.media.container import MediaItemContainer
 from program.updaters.trakt import Updater as Trakt, get_imdbid_from_tmdb, get_imdbid_from_tvdb
 
 
-class ListrrConfig(BaseModel):
-    enabled: bool
-    movie_lists: Optional[list]
-    show_lists: Optional[list]
-    api_key: Optional[str]
-    update_interval: int  # in seconds
-
-
 class Listrr:
     """Content class for Listrr"""
 
     def __init__(self, media_items: MediaItemContainer):
         self.key = "listrr"
         self.url = "https://listrr.pro/api"
-        self.settings = ListrrConfig(**settings_manager.get(f"content.{self.key}"))
+        self.settings = settings_manager.settings.content.listrr
         self.headers = {"X-Api-Key": self.settings.api_key}
         self.initialized = self.validate_settings()
         if not self.initialized:

backend/program/content/mdblist.py

@@ -1,24 +1,19 @@
 """Mdblist content module"""
 from typing import Optional
-from pydantic import BaseModel
-from utils.settings import settings_manager
+
+from program.settings.manager import settings_manager
 from utils.logger import logger
 from utils.request import RateLimitExceeded, RateLimiter, get, ping
 from program.media.container import MediaItemContainer
 from program.updaters.trakt import Updater as Trakt
 
 
-class MdblistConfig(BaseModel):
-    enabled: bool
-    api_key: Optional[str]
-    lists: Optional[list]
-
 class Mdblist:
     """Content class for mdblist"""
 
     def __init__(self, media_items: MediaItemContainer):
         self.key = "mdblist"
-        self.settings = MdblistConfig(**settings_manager.get(f"content.{self.key}"))
+        self.settings = settings_manager.settings.content.mdblist
         self.initialized = self.validate_settings()
         if not self.initialized:
             return

backend/program/content/overseerr.py

@@ -1,25 +1,20 @@
 """Mdblist content module"""
 from typing import Optional
-from pydantic import BaseModel
-from utils.settings import settings_manager
+
+
+from program.settings.manager import settings_manager
 from utils.logger import logger
 from utils.request import get, ping
 from program.media.container import MediaItemContainer
 from program.updaters.trakt import Updater as Trakt, get_imdbid_from_tmdb, get_imdbid_from_tvdb
 
 
-class OverseerrConfig(BaseModel):
-    enabled: bool
-    url: Optional[str]
-    api_key: Optional[str]
-
-
 class Overseerr:
     """Content class for overseerr"""
 
     def __init__(self, media_items: MediaItemContainer):
         self.key = "overseerr"
-        self.settings = OverseerrConfig(**settings_manager.get(f"content.{self.key}"))
+        self.settings = settings_manager.settings.content.overseerr
         self.headers = {"X-Api-Key": self.settings.api_key}
         self.initialized = self.validate_settings()
         if not self.initialized:

backend/program/content/plex_watchlist.py

@@ -1,18 +1,15 @@
 """Plex Watchlist Module"""
+import json
 from typing import Optional
-from pydantic import BaseModel
+
 from requests import ConnectTimeout, HTTPError
+
 from utils.request import get, ping
 from utils.logger import logger
-from utils.settings import settings_manager
+from program.settings.manager import settings_manager
 from program.media.container import MediaItemContainer
 from program.updaters.trakt import Updater as Trakt
-import json
-
 
-class PlexWatchlistConfig(BaseModel):
-    enabled: bool
-    rss: Optional[str]
 
 
 class PlexWatchlist:
@@ -21,11 +18,11 @@ class PlexWatchlist:
     def __init__(self, media_items: MediaItemContainer):
         self.key = "plex_watchlist"
         self.rss_enabled = False
-        self.settings = PlexWatchlistConfig(**settings_manager.get(f"content.{self.key}"))
+        self.settings = settings_manager.settings.content.plex_watchlist
         self.initialized = self.validate_settings()
         if not self.initialized:
             return
-        self.token = settings_manager.get("plex.token")
+        self.token = settings_manager.settings.plex.token
         self.media_items = media_items
         self.prev_count = 0
         self.updater = Trakt()

backend/program/content/trakt.py

@@ -1,30 +1,22 @@
 """Mdblist content module"""
 from time import time
 from typing import Optional
-from pydantic import BaseModel
-from utils.settings import settings_manager
+
+from program.settings.manager import settings_manager
 from utils.logger import logger
 from utils.request import get, ping
 from program.media.container import MediaItemContainer
 from program.updaters.trakt import Updater as Trakt, CLIENT_ID
 
 
-class TraktConfig(BaseModel):
-    enabled: bool
-    watchlist: Optional[list]
-    collection: Optional[list]
-    user_lists: Optional[list]
-    api_key: Optional[str]
-    update_interval: int  # in seconds
-
 
 class Trakt:
     """Content class for Trakt"""
 
     def __init__(self, media_items: MediaItemContainer):
         self.key = "trakt"
         self.url = None
-        self.settings = TraktConfig(**settings_manager.get(f"content.{self.key}"))
+        self.settings = settings_manager.settings.content.trakt
         self.headers = {"X-Api-Key": self.settings.api_key}
         self.initialized = self.validate_settings()
         if not self.initialized:

backend/program/plex.py

@@ -6,12 +6,12 @@
 import uuid
 from datetime import datetime
 from typing import Optional
+
 from plexapi.server import PlexServer
 from plexapi.exceptions import BadRequest, Unauthorized
-from pydantic import BaseModel
-# from program.updaters.trakt import get_imdbid_from_tvdb
+
 from utils.logger import logger
-from utils.settings import settings_manager as settings
+from program.settings.manager import settings_manager
 from program.media.container import MediaItemContainer
 from program.media.state import Symlink, Library
 from utils.request import get, post
@@ -23,12 +23,6 @@
 )
 
 
-class PlexConfig(BaseModel):
-    user: Optional[str] = None
-    token: Optional[str] = None
-    url: Optional[str] = None
-
-
 class Plex(threading.Thread):
     """Plex library class"""
 
@@ -37,12 +31,12 @@ def __init__(self, media_items: MediaItemContainer):
         self.key = "plex"
         self.initialized = False
         self.library_path = os.path.abspath(
-            os.path.dirname(settings.get("symlink.container_path"))
+            os.path.dirname(settings_manager.settings.symlink.container_path)
         )
         self.last_fetch_times = {}
 
         try:
-            self.settings = PlexConfig(**settings.get(self.key))
+            self.settings = settings_manager.settings.plex
             self.plex = PlexServer(
                 self.settings.url, self.settings.token, timeout=60
             )
@@ -185,7 +179,7 @@ def _oauth(self):
             additional_headers={
                 "X-Plex-Product": "Iceberg",
                 "X-Plex-Client-Identifier": random_uuid,
-                "X-Plex-Token": settings.get("plex.token"),
+                "X-Plex-Token": settings_manager.settings.plex.token,
             },
         )
         if not response.ok:

backend/program/realdebrid.py

@@ -3,30 +3,25 @@
 from pathlib import Path
 import time
 from typing import Optional
-from pydantic import BaseModel
 from requests import ConnectTimeout
 from utils.logger import logger
 from utils.request import get, post, ping
-from utils.settings import settings_manager
+from program.settings.manager import settings_manager
 from utils.parser import parser
 
 
 WANTED_FORMATS = [".mkv", ".mp4", ".avi"]
 RD_BASE_URL = "https://api.real-debrid.com/rest/1.0"
 
 
-class DebridConfig(BaseModel):
-    api_key: Optional[str]
-
-
 class Debrid:
     """Real-Debrid API Wrapper"""
 
     def __init__(self, _):
         # Realdebrid class library is a necessity
         self.initialized = False
         self.key = "real_debrid"
-        self.settings = DebridConfig(**settings_manager.get(self.key))
+        self.settings = settings_manager.settings.real_debrid
         self.auth_headers = {"Authorization": f"Bearer {self.settings.api_key}"}
         self.running = False
         if not self._validate_settings():