feat: save entry suffix separately

tagstudio/src/core/constants.py

@@ -7,7 +7,6 @@
 TS_FOLDER_NAME: str = ".TagStudio"
 BACKUP_FOLDER_NAME: str = "backups"
 COLLAGE_FOLDER_NAME: str = "collages"
-LIBRARY_FILENAME: str = "ts_library.json"
 
 # TODO: Turn this whitelist into a user-configurable blacklist.
 IMAGE_TYPES: list[str] = [
@@ -122,13 +121,13 @@
     + SHORTCUT_TYPES
 )
 
-
 TAG_FAVORITE = 1
 TAG_ARCHIVED = 0
 
 
 class LibraryPrefs(Enum):
-    IS_EXCLUDE_LIST = True
-    EXTENSION_LIST: list[str] = [".json", ".xmp", ".aae"]
+    IS_EXCLUDE_LIST: bool = True
+    EXTENSION_LIST: list[str] = ["json", "xmp", "aae"]
     PAGE_SIZE: int = 500
-    DB_VERSION: int = 1
+    # increase in case of db breaking change (for now)
+    DB_VERSION: int = 2

tagstudio/src/core/driver.py

@@ -0,0 +1,40 @@
+from pathlib import Path
+
+import structlog
+from PySide6.QtCore import QSettings
+from src.core.constants import TS_FOLDER_NAME
+from src.core.enums import SettingItems
+from src.core.library.alchemy.library import LibraryStatus
+
+logger = structlog.get_logger(__name__)
+
+
+class DriverMixin:
+    settings: QSettings
+
+    def evaluate_path(self, open_path: str | None) -> LibraryStatus:
+        """Check if the path of library is valid."""
+        library_path: Path | None = None
+        if open_path:
+            library_path = Path(open_path)
+            if not library_path.exists():
+                logger.error("Path does not exist.", open_path=open_path)
+                return LibraryStatus(success=False, message="Path does not exist.")
+        elif self.settings.value(
+            SettingItems.START_LOAD_LAST, defaultValue=True, type=bool
+        ) and self.settings.value(SettingItems.LAST_LIBRARY):
+            library_path = Path(str(self.settings.value(SettingItems.LAST_LIBRARY)))
+            if not (library_path / TS_FOLDER_NAME).exists():
+                logger.error(
+                    "TagStudio folder does not exist.",
+                    library_path=library_path,
+                    ts_folder=TS_FOLDER_NAME,
+                )
+                self.settings.setValue(SettingItems.LAST_LIBRARY, "")
+                # dont consider this a fatal error, just skip opening the library
+                library_path = None
+
+        return LibraryStatus(
+            success=True,
+            library_path=library_path,
+        )

tagstudio/src/core/library/alchemy/fields.py

@@ -18,27 +18,27 @@ class BaseField(Base):
     __abstract__ = True
 
     @declared_attr
-    def id(cls) -> Mapped[int]:  # noqa: N805
+    def id(self) -> Mapped[int]:
         return mapped_column(primary_key=True, autoincrement=True)
 
     @declared_attr
-    def type_key(cls) -> Mapped[str]:  # noqa: N805
+    def type_key(self) -> Mapped[str]:
         return mapped_column(ForeignKey("value_type.key"))
 
     @declared_attr
-    def type(cls) -> Mapped[ValueType]:  # noqa: N805
-        return relationship(foreign_keys=[cls.type_key], lazy=False)  # type: ignore
+    def type(self) -> Mapped[ValueType]:
+        return relationship(foreign_keys=[self.type_key], lazy=False)  # type: ignore
 
     @declared_attr
-    def entry_id(cls) -> Mapped[int]:  # noqa: N805
+    def entry_id(self) -> Mapped[int]:
         return mapped_column(ForeignKey("entries.id"))
 
     @declared_attr
-    def entry(cls) -> Mapped[Entry]:  # noqa: N805
-        return relationship(foreign_keys=[cls.entry_id])  # type: ignore
+    def entry(self) -> Mapped[Entry]:
+        return relationship(foreign_keys=[self.entry_id])  # type: ignore
 
     @declared_attr
-    def position(cls) -> Mapped[int]:  # noqa: N805
+    def position(self) -> Mapped[int]:
         return mapped_column(default=0)
 
     def __hash__(self):

tagstudio/src/core/library/alchemy/library.py

@@ -48,8 +48,6 @@
 from .joins import TagField, TagSubtag
 from .models import Entry, Folder, Preferences, Tag, TagAlias, ValueType
 
-LIBRARY_FILENAME: str = "ts_library.sqlite"
-
 logger = structlog.get_logger(__name__)
 
 
@@ -115,6 +113,15 @@ def __getitem__(self, index: int) -> Entry:
         return self.items[index]
 
 
+@dataclass
+class LibraryStatus:
+    """Keep status of library opening operation."""
+
+    success: bool
+    library_path: Path | None = None
+    message: str | None = None
+
+
 class Library:
     """Class for the Library object, and all CRUD operations made upon it."""
 
@@ -123,30 +130,28 @@ class Library:
     engine: Engine | None
     folder: Folder | None
 
+    FILENAME: str = "ts_library.sqlite"
+
     def close(self):
         if self.engine:
             self.engine.dispose()
         self.library_dir = None
         self.storage_path = None
         self.folder = None
 
-    def open_library(self, library_dir: Path | str, storage_path: str | None = None) -> None:
-        if isinstance(library_dir, str):
-            library_dir = Path(library_dir)
-
-        self.library_dir = library_dir
+    def open_library(self, library_dir: Path, storage_path: str | None = None) -> LibraryStatus:
         if storage_path == ":memory:":
             self.storage_path = storage_path
         else:
-            self.verify_ts_folders(self.library_dir)
-            self.storage_path = self.library_dir / TS_FOLDER_NAME / LIBRARY_FILENAME
+            self.verify_ts_folders(library_dir)
+            self.storage_path = library_dir / TS_FOLDER_NAME / self.FILENAME
 
         connection_string = URL.create(
             drivername="sqlite",
             database=str(self.storage_path),
         )
 
-        logger.info("opening library", connection_string=connection_string)
+        logger.info("opening library", library_dir=library_dir, connection_string=connection_string)
         self.engine = create_engine(connection_string)
         with Session(self.engine) as session:
             make_tables(self.engine)
@@ -183,11 +188,30 @@ def open_library(self, library_dir: Path | str, storage_path: str | None = None)
                     logger.debug("ValueType already exists", field=field)
                     session.rollback()
 
+            db_version = session.scalar(
+                select(Preferences).where(Preferences.key == LibraryPrefs.DB_VERSION.name)
+            )
+            # if the db version is different, we cant proceed
+            if db_version.value != LibraryPrefs.DB_VERSION.value:
+                logger.error(
+                    "DB version mismatch",
+                    db_version=db_version.value,
+                    expected=LibraryPrefs.DB_VERSION.value,
+                )
+                # TODO - handle migration
+                return LibraryStatus(
+                    success=False,
+                    message=(
+                        "Library version mismatch.\n"
+                        f"Found: v{db_version.value}, expected: v{LibraryPrefs.DB_VERSION.value}"
+                    ),
+                )
+
             # check if folder matching current path exists already
-            self.folder = session.scalar(select(Folder).where(Folder.path == self.library_dir))
+            self.folder = session.scalar(select(Folder).where(Folder.path == library_dir))
             if not self.folder:
                 folder = Folder(
-                    path=self.library_dir,
+                    path=library_dir,
                     uuid=str(uuid4()),
                 )
                 session.add(folder)
@@ -196,6 +220,10 @@ def open_library(self, library_dir: Path | str, storage_path: str | None = None)
                 session.commit()
                 self.folder = folder
 
+        # everything is fine, set the library path
+        self.library_dir = library_dir
+        return LibraryStatus(success=True, library_path=library_dir)
+
     @property
     def default_fields(self) -> list[BaseField]:
         with Session(self.engine) as session:
@@ -324,13 +352,17 @@ def add_entries(self, items: list[Entry]) -> list[int]:
 
         with Session(self.engine) as session:
             # add all items
-            session.add_all(items)
-            session.flush()
 
-            new_ids = [item.id for item in items]
+            try:
+                session.add_all(items)
+                session.flush()
+            except IntegrityError:
+                session.rollback()
+                logger.exception("IntegrityError")
+                return []
 
+            new_ids = [item.id for item in items]
             session.expunge_all()
-
             session.commit()
 
         return new_ids
@@ -396,9 +428,9 @@ def search_library(
 
             if not search.id:  # if `id` is set, we don't need to filter by extensions
                 if extensions and is_exclude_list:
-                    statement = statement.where(Entry.path.notilike(f"%.{','.join(extensions)}"))
+                    statement = statement.where(Entry.suffix.notin_(extensions))
                 elif extensions:
-                    statement = statement.where(Entry.path.ilike(f"%.{','.join(extensions)}"))
+                    statement = statement.where(Entry.suffix.in_(extensions))
 
             statement = statement.options(
                 selectinload(Entry.text_fields),
@@ -770,7 +802,7 @@ def save_library_backup_to_disk(self) -> Path:
         target_path = self.library_dir / TS_FOLDER_NAME / BACKUP_FOLDER_NAME / filename
 
         shutil.copy2(
-            self.library_dir / TS_FOLDER_NAME / LIBRARY_FILENAME,
+            self.library_dir / TS_FOLDER_NAME / self.FILENAME,
             target_path,
         )
 

tagstudio/src/core/library/alchemy/models.py

@@ -120,6 +120,7 @@ class Entry(Base):
     folder: Mapped[Folder] = relationship("Folder")
 
     path: Mapped[Path] = mapped_column(PathType, unique=True)
+    suffix: Mapped[str] = mapped_column()
 
     text_fields: Mapped[list[TextField]] = relationship(
         back_populates="entry",
@@ -177,6 +178,8 @@ def __init__(
         self.path = path
         self.folder = folder
 
+        self.suffix = path.suffix.lstrip(".").lower()
+
         for field in fields:
             if isinstance(field, TextField):
                 self.text_fields.append(field)

tagstudio/src/core/library/json/library.py

@@ -299,6 +299,8 @@ def compressed_dict(self):
 class Library:
     """Class for the Library object, and all CRUD operations made upon it."""
 
+    FILENAME: str = "ts_library.json"
+
     def __init__(self) -> None:
         # Library Info =========================================================
         self.library_dir: Path = None

tagstudio/src/qt/modals/file_extension.py

@@ -104,7 +104,7 @@ def save(self):
         for i in range(self.table.rowCount()):
             ext = self.table.item(i, 0)
             if ext and ext.text().strip():
-                extensions.append(ext.text().strip().lower())
+                extensions.append(ext.text().strip().lstrip(".").lower())
 
         # save preference
         self.lib.set_prefs(LibraryPrefs.EXTENSION_LIST, extensions)