Add TinyDB (#15)

pyproject.toml

@@ -28,12 +28,14 @@ classifiers = [
 ]
 
 [tool.poetry.dependencies]
-python = "^3.10"
+python = ">=3.10,<3.13"
+docstring-inheritance = "2.2.0"
 mutagen = "1.47.0"
 platformdirs = "4.2.0"
 pydantic = "2.7.0"
 pygame = "2.5.2"
 pymongo = "4.6.2"
+tinydb = "4.8.0"
 typer = "0.12.3"
 wurlitzer = "3.0.3"
 

rolabesti/conf/settings.py

@@ -8,10 +8,11 @@
 """
 
 import configparser
+import os
 import sys
 from os.path import exists
 
-from platformdirs import user_config_path, user_documents_dir, user_music_dir
+from platformdirs import user_config_path, user_data_path, user_documents_dir, user_music_dir
 
 from rolabesti import __app_name__
 
@@ -24,20 +25,28 @@
 OVERLAP_LENGTH = 3
 MUSIC_DIR = user_music_dir()
 COPY_DIR = user_documents_dir()
+DB = "tiny"
 
 # MongoDB.
 MONGO_HOST = "localhost"
 MONGO_PORT = 27017
 MONGO_DBNAME = "rolabesti"
 MONGO_COLNAME = "tracks"
 
+# TinyDB.
+TINY_DIR = user_data_path(__app_name__)
+TINY_FILE = TINY_DIR / "tracks.json"
+
+if DB == "tiny" and not exists(TINY_DIR):
+    os.mkdir(TINY_DIR)
+
 # Override settings
 conf_file = user_config_path(__app_name__) / f"{__app_name__}.conf"
 
 
 if exists(conf_file):
-    SETTINGS = ("MAX_TRACK_LENGTH", "MIN_TRACK_LENGTH", "MONGO_HOST", "MONGO_PORT", "MONGO_DBNAME", "MONGO_COLNAME",
-                "MUSIC_DIR", "PLAYER", "OVERLAP_LENGTH", "MAX_TRACKLIST_LENGTH", "SORTING", "COPY_DIR")
+    SETTINGS = ("MAX_TRACK_LENGTH", "MIN_TRACK_LENGTH", "MAX_TRACKLIST_LENGTH", "SORTING", "OVERLAP_LENGTH",
+                "MUSIC_DIR", "COPY_DIR", "DB", "MONGO_HOST", "MONGO_PORT", "MONGO_DBNAME", "MONGO_COLNAME")
     config = configparser.ConfigParser()
     config.read(conf_file)
 

rolabesti/controllers/controller.py

@@ -1,13 +1,18 @@
 from abc import ABC, abstractmethod
 
-from rolabesti.database import MongoDB
+from rolabesti.conf.settings import DB
+from rolabesti.database import MongoDB, TinyDB
 from rolabesti.logger import Logger
 
 
 class Controller(ABC):
     def __init__(self, parameters: dict) -> None:
         self.parameters = parameters
-        self.db = MongoDB()
+        match DB:
+            case "mongo":
+                self.db = MongoDB()
+            case "tiny":
+                self.db = TinyDB()
         self.logger = Logger()
 
     @abstractmethod

rolabesti/controllers/init.py

@@ -4,23 +4,31 @@
 from .parser import Parser
 
 
-COUNTS = (5, 10, 50, 100, 500, 1000, 2000, 5000, 10000, 15000, 20000, 25000, 30000, 35000, 40000, 45000, 50000)
+BATCH_SIZE = 100
+COUNTS = (100, 500, 1000, 2000, 5000, 10000, 15000, 20000, 25000, 30000, 35000, 40000, 45000, 50000)
 
 
 class InitController(Controller):
     def __call__(self) -> None:
         """Traverse directories to parse and insert tracks in database."""
-        self.logger.log(f"[green]initializing database with metadata of mp3 tracks "
-                        f"located at[/green] [blue]{self.parameters['music_directory']}[/blue]")
+        self.logger.log(f"[green]initializing database with metadata of mp3 tracks located at[/green] "
+                        f"[blue]{self.parameters['music_directory']}[/blue]")
 
         self.db.empty()
+        tracks = []
         count = 0
         parser = Parser()
         for trackpath in Path(self.parameters["music_directory"]).glob("**/*.[mM][pP]3"):
             if track := parser.parse(trackpath):
-                self.db.insert_one(track.model_dump(exclude_none=True))
+                tracks.append(track.model_dump(exclude_none=True))
                 count += 1
-                if count in COUNTS:
-                    self.logger.log(f"[yellow]{count}[/yellow] [green]tracks loaded so far[/green]")
+                if count % BATCH_SIZE == 0:
+                    self.db.insert_many(tracks)
+                    tracks.clear()
+                    if count in COUNTS:
+                        self.logger.log(f"[yellow]{count}[/yellow] [green]tracks loaded so far[/green]")
+
+        if tracks:
+            self.db.insert_many(tracks)
 
         self.logger.log(f"[yellow]{count}[/yellow] [green]track{'s'[:count!=1]} loaded in total[/green]")

rolabesti/database/__init__.py

@@ -1 +1,2 @@
 from .mongodb import MongoDB
+from .tinydb import TinyDB

rolabesti/database/db.py

@@ -1,25 +1,37 @@
-from abc import ABC, abstractmethod
+from abc import ABCMeta, abstractmethod
 from collections.abc import Generator
 from pathlib import Path
 
+from docstring_inheritance import NumpyDocstringInheritanceMeta
 
-class DB(ABC):
+
+# Support docstring inheritance.
+class ABCNumpyDocstringInheritanceMeta(ABCMeta, NumpyDocstringInheritanceMeta):
+    pass
+
+
+class DB(metaclass=ABCNumpyDocstringInheritanceMeta):
     @abstractmethod
     def count(self) -> int:
+        """Return the number of tracks."""
         pass
 
     @abstractmethod
     def empty(self) -> None:
+        """Delete all tracks."""
         pass
 
     @abstractmethod
-    def insert_one(self, track: dict) -> None:
+    def insert_many(self, tracks: list[dict]) -> None:
+        """Insert multiple tracks."""
         pass
 
     @abstractmethod
     def search(self, search_filters: dict) -> Generator[dict, None, None]:
+        """Return an iterator of matching tracks based on the search filters."""
         pass
 
     @abstractmethod
     def update_one(self, path: Path, field: str, value: str | int) -> None:
+        """Update one field in one track."""
         pass

rolabesti/database/tinydb.py

@@ -0,0 +1,54 @@
+import re
+from collections.abc import Generator
+from pathlib import Path
+
+from tinydb import Query, TinyDB as BaseTinyDB, where
+
+from .db import DB
+from rolabesti.conf.settings import TINY_FILE
+from rolabesti.models import FIELD_FILTERS
+
+
+class TinyDB(DB):
+    def __init__(self) -> None:
+        self.db = BaseTinyDB(TINY_FILE)
+
+    def count(self) -> int:
+        return len(self.db.all())
+
+    def empty(self) -> None:
+        self.db.truncate()
+
+    def insert_many(self, tracks: list[dict]) -> None:
+        self.db.insert_multiple(tracks)
+
+    def search(self, search_filters: dict) -> Generator[dict, None, None]:
+        tinydb_query = self._get_tinydb_query(search_filters)
+        for track in self.db.search(tinydb_query):
+            yield track
+
+    def update_one(self, path: Path, field: str, value: str | int) -> None:
+        self.db.update({field: value}, Query()["path"] == str(path))
+
+    @staticmethod
+    def _get_tinydb_query(search_filters: dict) -> Query:
+        """Get TinyDB query from search filters."""
+        query = Query().noop()
+
+        # Get length queries.
+        if (max_track_length := search_filters.get("max_track_length")) is not None:
+            query = query & (where('length') <= max_track_length)
+        if (min_track_length := search_filters.get("min_track_length")) is not None:
+            query = query & (where('length') >= min_track_length)
+
+        # Get field queries.
+        for field_filter in set.intersection(set(FIELD_FILTERS), set(search_filters)):
+            filter_value = search_filters[field_filter]
+            fields = FIELD_FILTERS[field_filter]
+            if len(fields) == 1:
+                query = query & where(fields[0]).search(filter_value, flags=re.IGNORECASE)
+            else:
+                query = query & (where(fields[0]).search(filter_value, flags=re.IGNORECASE) |
+                                 where(fields[1]).search(filter_value, flags=re.IGNORECASE))
+
+        return query