refactor API search code

Author: tkeriven

app/manager/mail/ClientImap.py

@@ -16,7 +16,7 @@
 from app.manager.mail.ClientMailServer import ClientMailServer
 from app.utils import errors as err
 from app.utils import constants as cs
-from app.utils.strings import quote, imap_join_folders
+from app.utils.strings import quote, imap_join_folders, escape_imap_string
 
 # Maximum debug output from imaplib
 #TODO all imap are logged, including login/auth password used SecretString (on ldap branch not in develoope now)
@@ -1978,6 +1978,75 @@ def delete_mail_permanently_from_folder_type(self, folder_type: str, mail_uid: s
         folder_path = self.folders_map_type_to_name[folder_type]
         self.delete_mails_by_uid(folder_path, mail_uid, move_to_trash=False, permanently=True)
 
+    def build_search_criteria(self, search_params: dict, include_deleted: bool) -> str:
+        """Build an IMAP SEARCH criteria string from the generic search_params dict.
+
+        :param search_params: Validated search parameters (from MailboxSearchSchema).
+        :type search_params: dict
+        :param include_deleted: Whether mails flagged \\Deleted should be included.
+        :type include_deleted: bool
+        :raises RequestException: If a date value cannot be parsed.
+        :return: IMAP SEARCH criteria string (e.g. "(NOT DELETED SUBJECT \"foo\")" or "ALL").
+        :rtype: str
+        """
+        criteria_parts: list[str] = []
+        if not include_deleted:
+            criteria_parts.append("NOT DELETED")
+
+        if search_params.get("text"):
+            criteria_parts.append(f'TEXT "{search_params["text"]}"')
+
+        if search_params.get("from_"):
+            escaped = escape_imap_string(search_params["from_"])
+            criteria_parts.append(f'FROM "{escaped}"')
+
+        if search_params.get("to"):
+            for addr in search_params["to"]:
+                criteria_parts.append(f'TO "{addr}"')
+
+        if search_params.get("subject"):
+            criteria_parts.append(f'SUBJECT "{search_params["subject"]}"')
+
+        if search_params.get("is_read") is True:
+            criteria_parts.append("SEEN")
+        elif search_params.get("is_read") is False:
+            criteria_parts.append("UNSEEN")
+
+        if search_params.get("is_flagged") is True:
+            criteria_parts.append("FLAGGED")
+        elif search_params.get("is_flagged") is False:
+            criteria_parts.append("UNFLAGGED")
+
+        if search_params.get("has_attachment") is True:
+            criteria_parts.append('HEADER Content-Type "multipart/mixed"')
+
+        if search_params.get("labels"):
+            for label in search_params["labels"]:
+                criteria_parts.append(f'KEYWORD "{label}"')
+
+        if search_params.get("date_range"):
+            date_range = search_params["date_range"]
+            if date_range.get("start"):
+                try:
+                    dt = datetime.fromisoformat(date_range["start"].replace("Z", "+00:00"))
+                    criteria_parts.append(f'SINCE {dt.strftime("%d-%b-%Y")}')
+                except (ValueError, AttributeError) as exc:
+                    raise RequestException(
+                        f"Invalid start date: {date_range['start']}",
+                        err.ERROR_MAIL_SEARCH_INVALID_DATE
+                    ) from exc
+            if date_range.get("end"):
+                try:
+                    dt = datetime.fromisoformat(date_range["end"].replace("Z", "+00:00"))
+                    criteria_parts.append(f'BEFORE {dt.strftime("%d-%b-%Y")}')
+                except (ValueError, AttributeError) as exc:
+                    raise RequestException(
+                        f"Invalid end date: {date_range['end']}",
+                        err.ERROR_MAIL_SEARCH_INVALID_DATE
+                    ) from exc
+
+        return "(" + " ".join(criteria_parts) + ")" if criteria_parts else "ALL"
+
     def _search_uids_in_folder(self, folder_path: str, criteria: str) -> str | None:
         """Execute an IMAP SEARCH in a single folder and return the UID set string, or None if no results.
 

app/manager/mail/ClientMailServer.py

@@ -62,6 +62,28 @@ def parse_fields_param(fields: str | None, fields_action: str | None) -> dict[st
 
         return {"with_content": with_content, "include_deleted": include_deleted}
 
+    @abstractmethod
+    def build_search_criteria(self, search_params: dict, include_deleted: bool) -> Any:
+        """Build a protocol-specific search criteria object/string from the generic,
+        protocol-agnostic ``search_params`` dict (as validated by MailboxSearchSchema).
+
+        This keeps every bit of protocol-specific search syntax (IMAP SEARCH syntax,
+        JMAP filter objects, ...) confined to the concrete client implementation, so
+        that callers (e.g. ModuleMail) stay protocol agnostic.
+
+        :param search_params: Validated search parameters (from MailboxSearchSchema),
+            with keys like "text", "from_", "to", "subject", "is_read", "is_flagged",
+            "has_attachment", "labels", "date_range".
+        :type search_params: dict
+        :param include_deleted: Whether mails flagged as deleted should be included.
+        :type include_deleted: bool
+        :raises RequestException: If a value in search_params cannot be translated
+            (e.g. an invalid date).
+        :return: A protocol-specific criteria value to pass to search_mails_with_content
+            / search_mails_without_content.
+        :rtype: Any
+        """
+
     @abstractmethod
     def connect(self) -> None:
         """Connect to the mail server."""
@@ -390,7 +412,18 @@ def save_draft(self, message: EmailMessage, uid: str | None = None) -> dict[str,
 
     @abstractmethod
     def get_quota(self) -> dict[str, Any] | None:
-        """Get quota information for the mailbox."""
+        """Get quota information for the mailbox.
+
+        Uses the IMAP GETQUOTAROOT command on the inbox folder.
+        Returns None if the server does not support QUOTA or the command is unavailable.
+
+        :return: Dictionary containing quota info, or None if unavailable:
+            {
+                "storage_used": int,   # storage used in KB
+                "storage_limit": int,  # storage limit in KB (0 if unlimited)
+            }
+        :rtype: dict[str, Any] | None
+        """
 
     @abstractmethod
     def search_mails_without_content(self, folders: list[str], criteria: str) -> Iterator[tuple[str, dict]]:

app/module/mail/ModuleMail.py

@@ -4,7 +4,6 @@
 import email as email_existing
 import email.mime.text
 import email.policy
-from datetime import datetime
 from email.header import decode_header, make_header
 from email.message import EmailMessage
 from email.message import Message
@@ -23,7 +22,7 @@
 from app.utils.maths.crypto_utils import decrypt_password
 from app.utils.module.importManager import import_and_instantiate_manager
 from app.utils.logger.logger import logger_mail_server
-from app.utils.strings import get_imap_config_from_url, get_domain_from_mail, get_domain_from_contact, escape_imap_string
+from app.utils.strings import get_imap_config_from_url, get_domain_from_mail, get_domain_from_contact
 from app.utils.constants import DELETE_MAIL_BEHAVIOR_MAP
 
 if TYPE_CHECKING:
@@ -717,9 +716,9 @@ def get_folder_mails(self, account_id: str, folder_name: str, collection_param:
     def search_mails(self, account_id: str, search_params: dict, collection_param: CollectionPaginateArgs) -> tuple[list[dict[str, Any]], int]:
         """Execute an advanced search across one or multiple folders.
 
-        Builds an IMAP SEARCH criteria string from ``search_params``, queries
-        each requested folder and returns a paginated list of matching mails together
-        with the total count.
+        Delegates the building of the protocol-specific search criteria (IMAP SEARCH
+        syntax, JMAP filter, ...) to the mail client, queries each requested folder
+        and returns a paginated list of matching mails together with the total count.
 
         :param account_id: The account identifier.
         :type account_id: str
@@ -729,7 +728,7 @@ def search_mails(self, account_id: str, search_params: dict, collection_param: C
         :type collection_param: CollectionPaginateArgs
         :return: A tuple of (list of mail dicts, total count).
         :rtype: tuple[list[dict[str, Any]], int]
-        :raises RequestException: If IMAP operations fail or dates are invalid.
+        :raises RequestException: If mail server operations fail or search_params are invalid.
         """
         client = self._open_client_for(account_id)
 
@@ -738,64 +737,8 @@ def search_mails(self, account_id: str, search_params: dict, collection_param: C
         without_content = not fields_params["with_content"]
         include_deleted = fields_params["include_deleted"]
 
-        # --- Build IMAP SEARCH criteria ---
-        criteria_parts: list[str] = []
-        if not include_deleted:
-            criteria_parts.append("NOT DELETED")
-
-        if search_params.get("text"):
-            criteria_parts.append(f'TEXT "{search_params["text"]}"')
-
-        if search_params.get("from_"):
-            escaped = escape_imap_string(search_params["from_"])
-            criteria_parts.append(f'FROM "{escaped}"')
-
-        if search_params.get("to"):
-            for addr in search_params["to"]:
-                criteria_parts.append(f'TO "{addr}"')
-
-        if search_params.get("subject"):
-            criteria_parts.append(f'SUBJECT "{search_params["subject"]}"')
-
-        if search_params.get("is_read") is True:
-            criteria_parts.append("SEEN")
-        elif search_params.get("is_read") is False:
-            criteria_parts.append("UNSEEN")
-
-        if search_params.get("is_flagged") is True:
-            criteria_parts.append("FLAGGED")
-        elif search_params.get("is_flagged") is False:
-            criteria_parts.append("UNFLAGGED")
-
-        if search_params.get("has_attachment") is True:
-            criteria_parts.append('HEADER Content-Type "multipart/mixed"')
-
-        if search_params.get("labels"):
-            for label in search_params["labels"]:
-                criteria_parts.append(f'KEYWORD "{label}"')
-
-        if search_params.get("date_range"):
-            date_range = search_params["date_range"]
-            if date_range.get("start"):
-                try:
-                    dt = datetime.fromisoformat(date_range["start"].replace("Z", "+00:00"))
-                    criteria_parts.append(f'SINCE {dt.strftime("%d-%b-%Y")}')
-                except (ValueError, AttributeError) as exc:
-                    raise RequestException(
-                        f"Invalid start date: {date_range['start']}",
-                        err.ERROR_MAIL_SEARCH_INVALID_DATE
-                    ) from exc
-            if date_range.get("end"):
-                try:
-                    dt = datetime.fromisoformat(date_range["end"].replace("Z", "+00:00"))
-                    criteria_parts.append(f'BEFORE {dt.strftime("%d-%b-%Y")}')
-                except (ValueError, AttributeError) as exc:
-                    raise RequestException(
-                        f"Invalid end date: {date_range['end']}",
-                        err.ERROR_MAIL_SEARCH_INVALID_DATE
-                    ) from exc
-
-        criteria = "(" + " ".join(criteria_parts) + ")" if criteria_parts else "ALL"
+        # --- Build the protocol-specific search criteria (delegated to the client) ---
+        criteria = client.build_search_criteria(search_params, include_deleted)
 
         # --- Determine folders to search ---
         folder_list = search_params.get("folders") or []