feat: Implement Query lang

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

@@ -1,7 +1,10 @@
 import enum
-from dataclasses import dataclass
+from dataclasses import dataclass, replace
 from pathlib import Path
 
+from src.core.query_lang import AST as Query  # noqa: N811
+from src.core.query_lang import Constraint, ConstraintType, Parser
+
 
 class TagColor(enum.IntEnum):
     DEFAULT = 1
@@ -50,13 +53,6 @@ def get_color_from_str(color_name: str) -> "TagColor":
         return TagColor.DEFAULT
 
 
-class SearchMode(enum.IntEnum):
-    """Operational modes for item searching."""
-
-    AND = 0
-    OR = 1
-
-
 class ItemType(enum.Enum):
     ENTRY = 0
     COLLATION = 1
@@ -68,71 +64,12 @@ class FilterState:
     """Represent a state of the Library grid view."""
 
     # these should remain
-    page_index: int | None = None
-    page_size: int | None = None
-    search_mode: SearchMode = SearchMode.AND  # TODO - actually implement this
+    page_index: int | None = 0
+    page_size: int | None = 500
 
     # these should be erased on update
-    # tag name
-    tag: str | None = None
-    # tag ID
-    tag_id: int | None = None
-
-    # entry id
-    id: int | None = None
-    # whole path
-    path: Path | str | None = None
-    # file name
-    name: str | None = None
-    # file type
-    filetype: str | None = None
-    mediatype: str | None = None
-
-    # a generic query to be parsed
-    query: str | None = None
-
-    def __post_init__(self):
-        # strip values automatically
-        if query := (self.query and self.query.strip()):
-            # parse the value
-            if ":" in query:
-                kind, _, value = query.partition(":")
-                value = value.replace('"', "")
-            else:
-                # default to tag search
-                kind, value = "tag", query
-
-            if kind == "tag_id":
-                self.tag_id = int(value)
-            elif kind == "tag":
-                self.tag = value
-            elif kind == "path":
-                self.path = value
-            elif kind == "name":
-                self.name = value
-            elif kind == "id":
-                self.id = int(self.id) if str(self.id).isnumeric() else self.id
-            elif kind == "filetype":
-                self.filetype = value
-            elif kind == "mediatype":
-                self.mediatype = value
-
-        else:
-            self.tag = self.tag and self.tag.strip()
-            self.tag_id = int(self.tag_id) if str(self.tag_id).isnumeric() else self.tag_id
-            self.path = self.path and str(self.path).strip()
-            self.name = self.name and self.name.strip()
-            self.id = int(self.id) if str(self.id).isnumeric() else self.id
-
-        if self.page_index is None:
-            self.page_index = 0
-        if self.page_size is None:
-            self.page_size = 500
-
-    @property
-    def summary(self):
-        """Show query summary."""
-        return self.query or self.tag or self.name or self.tag_id or self.path or self.id
+    # Abstract Syntax Tree Of the current Search Query
+    ast: Query = None
 
     @property
     def limit(self):
@@ -142,6 +79,37 @@ def limit(self):
     def offset(self):
         return self.page_size * self.page_index
 
+    @classmethod
+    def show_all(cls) -> "FilterState":
+        return FilterState()
+
+    @classmethod
+    def from_search_query(cls, search_query: str) -> "FilterState":
+        return cls(ast=Parser(search_query).parse())
+
+    @classmethod
+    def from_tag_id(cls, tag_id: int | str) -> "FilterState":
+        return cls(ast=Constraint(ConstraintType.TagID, str(tag_id), []))
+
+    @classmethod
+    def from_path(cls, path: Path | str) -> "FilterState":
+        return cls(ast=Constraint(ConstraintType.Path, str(path).strip(), []))
+
+    @classmethod
+    def from_mediatype(cls, mediatype: str) -> "FilterState":
+        return cls(ast=Constraint(ConstraintType.MediaType, mediatype, []))
+
+    @classmethod
+    def from_filetype(cls, filetype: str) -> "FilterState":
+        return cls(ast=Constraint(ConstraintType.FileType, filetype, []))
+
+    @classmethod
+    def from_tag_name(cls, tag_name: str) -> "FilterState":
+        return cls(ast=Constraint(ConstraintType.Tag, tag_name, []))
+
+    def with_page_size(self, page_size: int) -> "FilterState":
+        return replace(self, page_size=page_size)
+
 
 class FieldTypeEnum(enum.Enum):
     TEXT_LINE = "Text Line"

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

@@ -28,7 +28,6 @@
 from sqlalchemy.exc import IntegrityError
 from sqlalchemy.orm import (
     Session,
-    aliased,
     contains_eager,
     make_transient,
     selectinload,
@@ -42,7 +41,6 @@
     TS_FOLDER_NAME,
 )
 from ...enums import LibraryPrefs
-from ...media_types import MediaCategories
 from .db import make_tables
 from .enums import FieldTypeEnum, FilterState, TagColor
 from .fields import (
@@ -54,6 +52,7 @@
 )
 from .joins import TagField, TagSubtag
 from .models import Entry, Folder, Preferences, Tag, TagAlias, ValueType
+from .visitors import SQLBoolExpressionBuilder
 
 logger = structlog.get_logger(__name__)
 
@@ -402,6 +401,29 @@ def get_entry(self, entry_id: int) -> Entry | None:
             make_transient(entry)
             return entry
 
+    def get_entry_full(self, entry_id: int) -> Entry | None:
+        """Load entry an join with all joins and all tags."""
+        with Session(self.engine) as session:
+            statement = select(Entry).where(Entry.id == entry_id)
+            statement = (
+                statement.outerjoin(Entry.text_fields)
+                .outerjoin(Entry.datetime_fields)
+                .outerjoin(Entry.tag_box_fields)
+            )
+            statement = statement.options(
+                selectinload(Entry.text_fields),
+                selectinload(Entry.datetime_fields),
+                selectinload(Entry.tag_box_fields)
+                .joinedload(TagBoxField.tags)
+                .options(selectinload(Tag.aliases), selectinload(Tag.subtags)),
+            )
+            entry = session.scalar(statement)
+            if not entry:
+                return None
+            session.expunge(entry)
+            make_transient(entry)
+            return entry
+
     @property
     def entries_count(self) -> int:
         with Session(self.engine) as session:
@@ -518,63 +540,18 @@ def search_library(
         with Session(self.engine, expire_on_commit=False) as session:
             statement = select(Entry)
 
-            if search.tag:
-                SubtagAlias = aliased(Tag)  # noqa: N806
-                statement = (
-                    statement.join(Entry.tag_box_fields)
-                    .join(TagBoxField.tags)
-                    .outerjoin(Tag.aliases)
-                    .outerjoin(SubtagAlias, Tag.subtags)
-                    .where(
-                        or_(
-                            Tag.name.ilike(search.tag),
-                            Tag.shorthand.ilike(search.tag),
-                            TagAlias.name.ilike(search.tag),
-                            SubtagAlias.name.ilike(search.tag),
-                        )
-                    )
-                )
-            elif search.tag_id:
-                statement = (
-                    statement.join(Entry.tag_box_fields)
-                    .join(TagBoxField.tags)
-                    .where(Tag.id == search.tag_id)
-                )
-
-            elif search.id:
-                statement = statement.where(Entry.id == search.id)
-            elif search.name:
-                statement = select(Entry).where(
-                    and_(
-                        Entry.path.ilike(f"%{search.name}%"),
-                        # dont match directory name (ie. has following slash)
-                        ~Entry.path.ilike(f"%{search.name}%/%"),
-                    )
-                )
-            elif search.path:
-                search_str = str(search.path).replace("*", "%")
-                statement = statement.where(Entry.path.ilike(search_str))
-            elif search.filetype:
-                statement = statement.where(Entry.suffix.ilike(f"{search.filetype}"))
-            elif search.mediatype:
-                extensions: set[str] = set[str]()
-                for media_cat in MediaCategories.ALL_CATEGORIES:
-                    if search.mediatype == media_cat.name:
-                        extensions = extensions | media_cat.extensions
-                        break
-                # just need to map it to search db - suffixes do not have '.'
-                statement = statement.where(
-                    Entry.suffix.in_(map(lambda x: x.replace(".", ""), extensions))
+            if search.ast:
+                statement = statement.outerjoin(Entry.tag_box_fields).where(
+                    SQLBoolExpressionBuilder(self).visit(search.ast)
                 )
 
             extensions = self.prefs(LibraryPrefs.EXTENSION_LIST)
             is_exclude_list = self.prefs(LibraryPrefs.IS_EXCLUDE_LIST)
 
-            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.suffix.notin_(extensions))
-                elif extensions:
-                    statement = statement.where(Entry.suffix.in_(extensions))
+            if extensions and is_exclude_list:
+                statement = statement.where(Entry.suffix.notin_(extensions))
+            elif extensions:
+                statement = statement.where(Entry.suffix.in_(extensions))
 
             statement = statement.options(
                 selectinload(Entry.text_fields),
@@ -584,6 +561,8 @@ def search_library(
                 .options(selectinload(Tag.aliases), selectinload(Tag.subtags)),
             )
 
+            statement = statement.distinct(Entry.id)
+
             query_count = select(func.count()).select_from(statement.alias("entries"))
             count_all: int = session.execute(query_count).scalar()
 
@@ -597,7 +576,7 @@ def search_library(
 
             res = SearchResult(
                 total_count=count_all,
-                items=list(session.scalars(statement).unique()),
+                items=list(session.scalars(statement)),
             )
 
             session.expunge_all()