OP#2552 : add advanced search API and add deleted mail option in paginate decorator

Author: tkeriven

app/api/v1/mail/ApiMailMailbox.py

@@ -8,6 +8,7 @@
 from app.interface.mail.InterfaceApiMailMailbox import InterfaceApiMailMailbox
 from app.utils.logger.logger import logger_api
 from app.utils.api.ApiBaseResponse import ApiBaseResponse
+from app.utils.api.paginate_sort_filter import collection_paginate, CustomPaginateResponse
 from app.api.v1.mail.schemas.mailbox import (
     MailboxCreateSchema,
     MailboxUpdateSchema,
@@ -18,11 +19,14 @@
     DelegationResponseSchema,
     MailboxPurgeSchema,
     MailboxPurgeResponseSchema,
+    MailboxSearchSchema,
+    MailboxSearchResponseSchema,
 )
 
 if TYPE_CHECKING:
     from app.config.settings.ProcessSetting import ProcessSetting
     from app.auth.User import User
+    from app.utils.api.paginate_sort_filter import CollectionPaginateArgs
 
 blp = Blueprint("Mail Account", __name__, url_prefix="/mailboxes")
 
@@ -155,3 +159,38 @@ def post(self, purge_data: dict, account_id: str) -> ResponseReturnValue:
         interface: InterfaceApiMailMailbox = g.inter
         return interface.purge_mailbox(account_id, purge_data)
 
+
+@blp.route("/<string:account_id>/search")
+class ApiMailBoxesAccountSearch(MethodView):
+    """
+    Resource: Advanced Mail Search
+    """
+    @blp.arguments(MailboxSearchSchema, example=MailboxSearchSchema.example(), error_status_code=400)
+    @blp.response(200, MailboxSearchResponseSchema)
+    @collection_paginate(blp, can_sort=True, sort_value_set={"date", "relevance", "sender", "subject", "size"},
+                         can_filter=True, filter_value_set={"contents", "deleted"})
+    def post(self, search_params: dict, collection_param: "CollectionPaginateArgs", account_id: str) -> CustomPaginateResponse:
+        """
+        Advanced mail search across one or multiple folders.
+
+        * **operator**: str, 'AND' (default) or 'OR' - how the criteria below are combined.
+          With 'AND' every provided criterion must match, with 'OR' at least one must match.
+        * **text**: str, full text search in subject/sender/recipients/body
+        * **folders**: list[str], list of folder paths to search in (e.g. ["INBOX", "Sent"] or ["all"] for all folders)
+        * **include_subfolders**: bool, default True - when True, also search the subfolders of each folder listed in "folders"; when False, search only the exact folders listed. Ignored when "folders" is empty or ["all"].
+        * **date_range**: dict, date range for the search (e.g. {"from": "2023-01-01", "to": "2023-01-31"})
+        * **has_attachments**: bool, whether to search for emails with attachments
+        * **to**: str, email address to search for in either the recipient (To) or copy (Cc) headers
+        * **bcc**: str, blind copy (Bcc) email address to search for
+        * **from**: list[str], list of sender email addresses to search for
+        * **subject** : str, keywords to search for in the email subject
+        * **attachment_type**: list[str], list of attachment types to search for (e.g. ["pdf", "jpg"])
+        * **is_read**: bool, whether to search for read or unread emails
+        * **labels**: list[str], list of labels/tags to search for
+
+        All search criteria are optional and combined using the "operator" field (AND by default, OR to match any criterion).
+        Pagination, sorting and field filtering are controlled via query parameters (page, page_size, sort_by, sort_order, fields, fields_action).
+        """
+        logger_api.debug("Calling ApiMailBoxesAccountSearch.post for account_id: %s with params: %s", account_id, search_params)
+        interface: InterfaceApiMailMailbox = g.inter
+        return interface.search_mailbox(account_id, search_params, collection_param)

app/api/v1/mail/schemas/mail.py

@@ -217,7 +217,7 @@ def filter_by_values() -> set:
         """
         return values available for sorting by
         """
-        return {"contents"}
+        return {"contents", "deleted"}
 
     @classmethod
     def example(cls) -> dict:

app/api/v1/mail/schemas/mailbox.py

@@ -604,3 +604,102 @@ def example(cls) -> dict:
             }
         }
 
+class DateRangeSchema(Schema):
+    """
+    Schema for date range filter in advanced search
+    """
+    start = fields.String(required=False, allow_none=True, metadata={"description": "Start date in ISO 8601 format (e.g. 2026-05-01T00:00:00Z)"})
+    end = fields.String(required=False, allow_none=True, metadata={"description": "End date in ISO 8601 format (e.g. 2026-05-19T23:59:59Z)"})
+
+    @classmethod
+    def example(cls) -> dict:
+        return {
+            "start": "2026-05-01T00:00:00Z",
+            "end": "2026-05-19T23:59:59Z"
+        }
+
+
+class MailboxSearchSchema(Schema):
+    """
+    Schema for POST /mailboxes/<account_id>/search - Advanced mail search.
+
+    All fields are optional. When multiple criteria are provided, they are combined
+    using the "operator" field: "AND" (default, every criterion must match) or
+    "OR" (at least one criterion must match).
+    """
+    operator = fields.String(
+        required=False,
+        allow_none=True,
+        load_default="AND",
+        validate=validate.OneOf(["AND", "OR"]),
+        metadata={"description": "Logical operator combining the search criteria below: 'AND' (default) requires every provided criterion to match, 'OR' requires at least one to match"}
+    )
+    text = fields.String(required=False, allow_none=True, load_default=None, metadata={"description": "Full-text search in body and headers"})
+    from_ = fields.String(required=False, allow_none=True, load_default=None, data_key="from", metadata={"description": "Filter by sender email address"})
+    to = fields.String(required=False, allow_none=True, load_default=None, metadata={"description": "Filter by recipient email address (matches either the To or the Cc header)"})
+    bcc = fields.String(required=False, allow_none=True, load_default=None, metadata={"description": "Filter by Bcc recipient email address"})
+    subject = fields.String(required=False, allow_none=True, load_default=None, metadata={"description": "Filter by subject (substring match)"})
+    has_attachment = fields.Boolean(required=False, allow_none=True, load_default=None, metadata={"description": "Filter mails that have (or don't have) attachments"})
+    attachment_type = fields.List(fields.String(), required=False, allow_none=True, load_default=None, metadata={"description": "Filter by attachment file extensions (e.g. ['pdf', 'jpg'])"})
+    date_range = fields.Nested(DateRangeSchema, required=False, allow_none=True, load_default=None, metadata={"description": "Filter by date range"})
+    is_read = fields.Boolean(required=False, allow_none=True, load_default=None, metadata={"description": "Filter by read/unread status"})
+    is_flagged = fields.Boolean(required=False, allow_none=True, load_default=None, metadata={"description": "Filter by starred (flagged) status"})
+    folders = fields.List(fields.String(), required=False, allow_none=True, load_default=None, metadata={"description": "Folders to search in (use ['all'] for entire mailbox)"})
+    include_subfolders = fields.Boolean(required=False, allow_none=True, load_default=True, metadata={"description": "If True (default), also search in the subfolders of each folder listed in 'folders'. If False, search only in the exact folders listed"})
+    labels = fields.List(fields.String(), required=False, allow_none=True, load_default=None, metadata={"description": "Filter by IMAP keyword labels"})
+
+    @classmethod
+    def example(cls) -> dict:
+        """Example data for advanced mail search.
+
+        :return: Example search payload
+        :rtype: dict
+        """
+        return {
+            "operator": "AND",
+            "text": "contrat urgent",
+            "from": "customer@entreprise.com",
+            "to": "jdoe@domaine.com",
+            "bcc": "hidden@domaine.com",
+            "subject": "Projet X",
+            "has_attachment": True,
+            "attachment_type": ["pdf", "jpg"],
+            "date_range": {
+                "start": "2025-05-01T00:00:00Z",
+                "end": "2026-05-19T23:59:59Z"
+            },
+            "is_read": False,
+            "is_flagged": True,
+            "folders": ["INBOX", "Archive"],
+            "include_subfolders": True,
+            "labels": ["important", "work"],
+        }
+
+
+class MailboxSearchResponseSchema(ApiBaseResponse):
+    """
+    Schema for the response of the advanced mail search endpoint.
+    """
+    data = fields.Dict(required=False, allow_none=True, metadata={"description": "Search results with mails list and total count"})
+
+    @classmethod
+    def example(cls) -> dict:
+        return {
+            "error_code": 0,
+            "error_msg": "",
+            "data": {
+                "total": 2,
+                "mails": [
+                    {
+                        "uid": "42",
+                        "subject": "Projet X - Contrat urgent",
+                        "from": {"name": "Client", "email": "client@entreprise.com"},
+                        "date": "Tue, 19 May 2026 10:00:00 +0000",
+                        "seen": False,
+                        "flagged": True,
+                        "has_attachment": True,
+                        "folder": "INBOX"
+                    }
+                ]
+            }
+        }

app/interface/mail/InterfaceApiMailMailbox.py

@@ -15,6 +15,7 @@
 if TYPE_CHECKING:
     from app.config.settings.ProcessSetting import ProcessSetting
     from app.auth.User import User
+    from app.utils.api.paginate_sort_filter import CollectionPaginateArgs
 
 
 class InterfaceApiMailMailbox:
@@ -296,3 +297,25 @@ def send_mail(self, account_id: str, mail_data: dict, draft_uid: str | None = No
                 logger_api.warning("Failed to delete draft mail uid %s for user %s, account %s: %s", draft_uid, self.user.uid, account_id, str(ex))
 
         return create_api_base_response(None)
+
+    def search_mailbox(self, account_id: str, search_params: dict, collection_param: "CollectionPaginateArgs") -> tuple[int, dict, int]:
+        """Advanced mail search across one or multiple folders for the given account.
+
+        :param account_id: The account identifier ("0" for main account)
+        :type account_id: str
+        :param search_params: Validated search parameters (from MailboxSearchSchema)
+        :type search_params: dict
+        :param collection_param: Pagination, sorting and filtering parameters.
+        :type collection_param: CollectionPaginateArgs
+        :return: A tuple of (total_count, API response dict, status code)
+        :rtype: tuple[int, dict, int]
+        """
+        if account_id != cs.DEFAULT_IDENTITY_KEY_VALUE and not self.user_module_settings.SOGO_D_ALLOW_EXT_MAIL_ACCOUNT:
+            return 0, *create_api_base_response(error=err.ERROR_EXTERNAL_ACCOUNT_FORBIDDEN)
+
+        try:
+            result, total = self.mail_module.search_mails(account_id, search_params, collection_param)
+        except RequestException as ex:
+            logger_api.error("Request exception in search_mailbox for user %s, account %s: %s", self.user.uid, account_id, str(ex))
+            return 0, *create_api_base_response(None, ex.error)
+        return total, *create_api_base_response(result)