feat: improvements

app/api/v1/gmail.py

@@ -1,7 +1,7 @@
 from __future__ import annotations
 
 import time
-from typing import Any, Dict, List, Optional
+from typing import Any, Dict, Optional
 
 from fastapi import APIRouter, Depends, Header, HTTPException
 
@@ -17,13 +17,16 @@ from app.models.schemas import (
     GmailDraftUpdateRequest,
     GmailFilterCreateRequest,
     GmailGenericResponse,
+    GmailImportMessageRequest,
     GmailInsertMessageRequest,
+    GmailLabelPatchRequest,
     GmailLabelRequest,
     GmailMessageModifyRequest,
     GmailRawMessageRequest,
     GmailRefreshResponse,
     GmailSendResponse,
     GmailSettingsUpdateRequest,
+    GmailWatchRequest,
 )
 from app.services.gmail_service import (
     GmailAPIError,
@@ -177,6 +180,77 @@ async def get_profile(
     return _ok(start, creds, data)
 
 
+# ---------------------------------------------------------------------------
+# Watch / Stop / History (push notifications & change history)
+# ---------------------------------------------------------------------------
+
+@router.post("/watch", response_model=GmailGenericResponse,
+             summary="Set up push notifications for mailbox changes (users.watch)")
+async def watch_mailbox(
+    body: GmailWatchRequest,
+    creds: GmailCredentials = Depends(_credentials),
+    service: GmailService = Depends(get_gmail_service),
+):
+    start = time.perf_counter()
+    try:
+        data = await service.watch(
+            creds,
+            user_id=body.user_id,
+            topic_name=body.topic_name,
+            label_ids=body.label_ids,
+            label_filter_action=body.label_filter_action,
+        )
+    except GmailAPIError as exc:
+        raise _http_error(exc) from exc
+    _logger.info("Gmail watch enabled for %s (%.2fms)", body.user_id, _elapsed_ms(start))
+    return _ok(start, creds, data)
+
+
+@router.post("/stop", response_model=GmailGenericResponse,
+             summary="Stop receiving push notifications (users.stop)")
+async def stop_watch_mailbox(
+    creds: GmailCredentials = Depends(_credentials),
+    service: GmailService = Depends(get_gmail_service),
+    user_id: Optional[str] = None,
+):
+    start = time.perf_counter()
+    try:
+        await service.stop_watch(creds, user_id=user_id or "me")
+    except GmailAPIError as exc:
+        raise _http_error(exc) from exc
+    _logger.info("Gmail watch stopped (%.2fms)", _elapsed_ms(start))
+    return _ok(start, creds, {"stopped": True})
+
+
+@router.get("/history", response_model=GmailGenericResponse,
+            summary="List mailbox change history (users.history.list)")
+async def list_history(
+    creds: GmailCredentials = Depends(_credentials),
+    service: GmailService = Depends(get_gmail_service),
+    start_history_id: int = ...,
+    label_id: Optional[str] = None,
+    history_types: Optional[str] = None,
+    max_results: Optional[int] = None,
+    page_token: Optional[str] = None,
+    user_id: Optional[str] = None,
+):
+    start = time.perf_counter()
+    try:
+        data = await service.list_history(
+            creds,
+            user_id=user_id or "me",
+            start_history_id=start_history_id,
+            label_id=label_id,
+            history_types=history_types.split(",") if history_types else None,
+            max_results=max_results,
+            page_token=page_token,
+        )
+    except GmailAPIError as exc:
+        raise _http_error(exc) from exc
+    _logger.info("Gmail history listed (%.2fms)", _elapsed_ms(start))
+    return _ok(start, creds, data)
+
+
 # ---------------------------------------------------------------------------
 # Messages
 # ---------------------------------------------------------------------------
@@ -362,6 +436,37 @@ async def insert_message(
     )
 
 
+@router.post("/messages/import", response_model=GmailSendResponse,
+             summary="Import an RFC 2822 message into the mailbox (users.messages.import)")
+async def import_message(
+    body: GmailImportMessageRequest,
+    creds: GmailCredentials = Depends(_credentials),
+    service: GmailService = Depends(get_gmail_service),
+):
+    start = time.perf_counter()
+    try:
+        data = await service.import_message(
+            creds,
+            body.raw,
+            user_id=_user_id(body.delegate),
+            label_ids=body.label_ids,
+            internal_date_source=body.internal_date_source,
+            never_mark_spam=body.never_mark_spam,
+            process_for_calendar=body.process_for_calendar,
+        )
+    except GmailAPIError as exc:
+        raise _http_error(exc) from exc
+    _logger.info("Gmail message imported (%.2fms)", _elapsed_ms(start))
+    return GmailSendResponse(
+        success=True,
+        time_ms=_elapsed_ms(start),
+        message_id=data.get("id"),
+        thread_id=data.get("threadId"),
+        refreshed_access_token=creds.refreshed_access_token,
+        data=data,
+    )
+
+
 @router.post("/messages/batchModify", response_model=GmailGenericResponse,
              summary="Add/remove labels across many messages")
 async def batch_modify(
@@ -802,6 +907,30 @@ async def update_label(
     return _ok(start, creds, data)
 
 
+@router.patch("/labels/{label_id}", response_model=GmailGenericResponse,
+              summary="Partially update a user label (users.labels.patch)")
+async def patch_label(
+    label_id: str,
+    body: GmailLabelPatchRequest,
+    creds: GmailCredentials = Depends(_credentials),
+    service: GmailService = Depends(get_gmail_service),
+    user_id: Optional[str] = None,
+):
+    start = time.perf_counter()
+    payload = body.model_dump(exclude_none=True)
+    if not payload:
+        raise HTTPException(
+            status_code=400,
+            detail="At least one label field (name, label_list_visibility, message_list_visibility, color) is required.",
+        )
+    try:
+        data = await service.patch_label(creds, label_id, payload, user_id=user_id or "me")
+    except GmailAPIError as exc:
+        raise _http_error(exc) from exc
+    _logger.info("Gmail label %s patched (%.2fms)", label_id, _elapsed_ms(start))
+    return _ok(start, creds, data)
+
+
 @router.delete("/labels/{label_id}", response_model=GmailGenericResponse,
                summary="Delete a user label")
 async def delete_label(
@@ -889,6 +1018,105 @@ async def update_vacation(
     return _ok(start, creds, data)
 
 
+@router.get("/settings/imap", response_model=GmailGenericResponse,
+            summary="Get IMAP settings")
+async def get_imap(
+    creds: GmailCredentials = Depends(_credentials),
+    service: GmailService = Depends(get_gmail_service),
+    user_id: Optional[str] = None,
+):
+    start = time.perf_counter()
+    try:
+        data = await service.get_imap(creds, user_id=user_id or "me")
+    except GmailAPIError as exc:
+        raise _http_error(exc) from exc
+    _logger.info("Gmail IMAP settings fetched (%.2fms)", _elapsed_ms(start))
+    return _ok(start, creds, data)
+
+
+@router.put("/settings/imap", response_model=GmailGenericResponse,
+            summary="Update IMAP settings")
+async def update_imap(
+    body: GmailSettingsUpdateRequest,
+    creds: GmailCredentials = Depends(_credentials),
+    service: GmailService = Depends(get_gmail_service),
+    user_id: Optional[str] = None,
+):
+    start = time.perf_counter()
+    try:
+        data = await service.update_imap(creds, body.payload, user_id=user_id or "me")
+    except GmailAPIError as exc:
+        raise _http_error(exc) from exc
+    _logger.info("Gmail IMAP settings updated (%.2fms)", _elapsed_ms(start))
+    return _ok(start, creds, data)
+
+
+@router.get("/settings/pop", response_model=GmailGenericResponse,
+            summary="Get POP settings")
+async def get_pop(
+    creds: GmailCredentials = Depends(_credentials),
+    service: GmailService = Depends(get_gmail_service),
+    user_id: Optional[str] = None,
+):
+    start = time.perf_counter()
+    try:
+        data = await service.get_pop(creds, user_id=user_id or "me")
+    except GmailAPIError as exc:
+        raise _http_error(exc) from exc
+    _logger.info("Gmail POP settings fetched (%.2fms)", _elapsed_ms(start))
+    return _ok(start, creds, data)
+
+
+@router.put("/settings/pop", response_model=GmailGenericResponse,
+            summary="Update POP settings")
+async def update_pop(
+    body: GmailSettingsUpdateRequest,
+    creds: GmailCredentials = Depends(_credentials),
+    service: GmailService = Depends(get_gmail_service),
+    user_id: Optional[str] = None,
+):
+    start = time.perf_counter()
+    try:
+        data = await service.update_pop(creds, body.payload, user_id=user_id or "me")
+    except GmailAPIError as exc:
+        raise _http_error(exc) from exc
+    _logger.info("Gmail POP settings updated (%.2fms)", _elapsed_ms(start))
+    return _ok(start, creds, data)
+
+
+@router.get("/settings/language", response_model=GmailGenericResponse,
+            summary="Get language settings")
+async def get_language(
+    creds: GmailCredentials = Depends(_credentials),
+    service: GmailService = Depends(get_gmail_service),
+    user_id: Optional[str] = None,
+):
+    start = time.perf_counter()
+    try:
+        data = await service.get_language(creds, user_id=user_id or "me")
+    except GmailAPIError as exc:
+        raise _http_error(exc) from exc
+    _logger.info("Gmail language settings fetched (%.2fms)", _elapsed_ms(start))
+    return _ok(start, creds, data)
+
+
+@router.put("/settings/language", response_model=GmailGenericResponse,
+            summary="Update language settings")
+async def update_language(
+    body: GmailSettingsUpdateRequest,
+    creds: GmailCredentials = Depends(_credentials),
+    service: GmailService = Depends(get_gmail_service),
+    user_id: Optional[str] = None,
+):
+    start = time.perf_counter()
+    try:
+        data = await service.update_language(creds, body.payload, user_id=user_id or "me")
+    except GmailAPIError as exc:
+        raise _http_error(exc) from exc
+    _logger.info("Gmail language settings updated (%.2fms)", _elapsed_ms(start))
+    return _ok(start, creds, data)
+
+
 @router.get("/settings/filters", response_model=GmailGenericResponse,
             summary="List all filters")
 async def list_filters(
@@ -1111,6 +1339,23 @@ async def delete_send_as(
     return _ok(start, creds, {"deleted": send_as_email})
 
 
+@router.post("/settings/send-as/{send_as_email}/verify", response_model=GmailGenericResponse,
+             summary="Send a verification email for a send-as alias (users.settings.sendAs.verify)")
+async def verify_send_as(
+    send_as_email: str,
+    creds: GmailCredentials = Depends(_credentials),
+    service: GmailService = Depends(get_gmail_service),
+    user_id: Optional[str] = None,
+):
+    start = time.perf_counter()
+    try:
+        await service.verify_send_as(creds, send_as_email, user_id=user_id or "me")
+    except GmailAPIError as exc:
+        raise _http_error(exc) from exc
+    _logger.info("Gmail send-as %s verification triggered (%.2fms)", send_as_email, _elapsed_ms(start))
+    return _ok(start, creds, {"verification_sent": send_as_email})
+
+
 @router.get("/settings/delegates", response_model=GmailGenericResponse,
             summary="List delegates")
 async def list_delegates(

app/models/schemas.py

@@ -1863,6 +1863,35 @@ class GmailLabelRequest(BaseModel):
     color: Optional[Dict[str, str]] = Field(None, description="Label color (background/text color hex values)")
 
 
+class GmailLabelPatchRequest(BaseModel):
+    """Partial update for a label (Gmail users.labels.patch)."""
+
+    name: Optional[str] = Field(None, min_length=1, max_length=255, description="New label name")
+    label_list_visibility: Optional[str] = Field(None, pattern="^(labelShow|labelHide|labelShowIfUnread)$", description="Label list visibility")
+    message_list_visibility: Optional[str] = Field(None, pattern="^(show|hide)$", description="Message list visibility")
+    color: Optional[Dict[str, str]] = Field(None, description="Label color (background/text color hex values)")
+
+
+class GmailWatchRequest(BaseModel):
+    """Set up push notifications for mailbox changes (users.watch)."""
+
+    topic_name: str = Field(..., min_length=1, description="Cloud Pub/Sub topic name, e.g. projects/{project}/topics/{topic}")
+    label_ids: Optional[List[str]] = Field(None, description="Only send notifications for messages with these label IDs")
+    label_filter_action: Optional[str] = Field(None, pattern="^(include|exclude)$", description="Whether labelIds are include or exclude filters")
+    user_id: str = Field("me", description="Mailbox to watch (default 'me')")
+
+
+class GmailImportMessageRequest(BaseModel):
+    """Import an RFC 2822 message directly into the mailbox (users.messages.import)."""
+
+    raw: str = Field(..., min_length=1, description="Base64url-encoded RFC 2822 message")
+    label_ids: Optional[List[str]] = Field(None, description="Label IDs to apply on import")
+    internal_date_source: Optional[str] = Field(None, pattern="^(dateHeader|receivedTime)$", description="Source for internal date")
+    never_mark_spam: bool = Field(False, description="Ignore 'X-Gmail-Labels' / spam classifier and never mark as spam")
+    process_for_calendar: bool = Field(False, description="Process the email for calendar events")
+    delegate: str = Field("me", description="Mailbox to act on (default 'me')")
+
+
 class GmailMessageAttachmentMeta(BaseModel):
     attachment_id: str = Field(..., description="Gmail attachment ID")
     filename: str = Field("", description="Attachment file name")

app/services/gmail_service.py

@@ -493,6 +493,30 @@ class GmailService:
             creds, "POST", f"/users/{user_id}/messages", json_body=body
         )
 
+    async def import_message(
+        self,
+        creds: GmailCredentials,
+        raw: str,
+        user_id: str = "me",
+        *,
+        label_ids: Optional[List[str]] = None,
+        internal_date_source: Optional[str] = None,
+        never_mark_spam: bool = False,
+        process_for_calendar: bool = False,
+    ) -> Dict[str, Any]:
+        body: Dict[str, Any] = {"raw": raw}
+        if label_ids:
+            body["labelIds"] = label_ids
+        if internal_date_source:
+            body["internalDateSource"] = internal_date_source
+        if never_mark_spam:
+            body["neverMarkSpam"] = True
+        if process_for_calendar:
+            body["processForCalendar"] = True
+        return await self._request(
+            creds, "POST", f"/users/{user_id}/messages/import", json_body=body
+        )
+
     async def modify_message(
         self,
         creds: GmailCredentials,
@@ -566,6 +590,55 @@ class GmailService:
             f"/users/{user_id}/messages/{message_id}/attachments/{attachment_id}",
         )
 
+    # ------------------------------------------------------------------
+    # Push notifications (watch / stop)
+    # ------------------------------------------------------------------
+
+    async def watch(
+        self,
+        creds: GmailCredentials,
+        user_id: str = "me",
+        *,
+        topic_name: str,
+        label_ids: Optional[List[str]] = None,
+        label_filter_action: Optional[str] = None,
+    ) -> Dict[str, Any]:
+        body: Dict[str, Any] = {"topicName": topic_name}
+        if label_ids:
+            body["labelIds"] = label_ids
+        if label_filter_action:
+            body["labelFilterAction"] = label_filter_action
+        return await self._request(creds, "POST", f"/users/{user_id}/watch", json_body=body)
+
+    async def stop_watch(self, creds: GmailCredentials, user_id: str = "me") -> None:
+        await self._request(creds, "POST", f"/users/{user_id}/stop")
+
+    # ------------------------------------------------------------------
+    # History
+    # ------------------------------------------------------------------
+
+    async def list_history(
+        self,
+        creds: GmailCredentials,
+        user_id: str = "me",
+        *,
+        start_history_id: int,
+        label_id: Optional[str] = None,
+        history_types: Optional[List[str]] = None,
+        max_results: Optional[int] = None,
+        page_token: Optional[str] = None,
+    ) -> Dict[str, Any]:
+        params: Dict[str, Any] = {"startHistoryId": start_history_id}
+        if label_id:
+            params["labelId"] = label_id
+        if history_types:
+            params["historyTypes"] = history_types
+        if max_results is not None:
+            params["maxResults"] = max_results
+        if page_token:
+            params["pageToken"] = page_token
+        return await self._request(creds, "GET", f"/users/{user_id}/history", params=params)
+
     # ------------------------------------------------------------------
     # Drafts
     # ------------------------------------------------------------------
@@ -767,6 +840,17 @@ class GmailService:
             creds, "PUT", f"/users/{user_id}/labels/{label_id}", json_body=body
         )
 
+    async def patch_label(
+        self,
+        creds: GmailCredentials,
+        label_id: str,
+        payload: Dict[str, Any],
+        user_id: str = "me",
+    ) -> Dict[str, Any]:
+        return await self._request(
+            creds, "PATCH", f"/users/{user_id}/labels/{label_id}", json_body=payload
+        )
+
     async def delete_label(
         self, creds: GmailCredentials, label_id: str, user_id: str = "me"
     ) -> None:
@@ -796,6 +880,36 @@ class GmailService:
             creds, "PUT", f"/users/{user_id}/settings/vacation", json_body=payload
         )
 
+    async def get_imap(self, creds: GmailCredentials, user_id: str = "me") -> Dict[str, Any]:
+        return await self._request(creds, "GET", f"/users/{user_id}/settings/imap")
+
+    async def update_imap(
+        self, creds: GmailCredentials, payload: Dict[str, Any], user_id: str = "me"
+    ) -> Dict[str, Any]:
+        return await self._request(
+            creds, "PUT", f"/users/{user_id}/settings/imap", json_body=payload
+        )
+
+    async def get_pop(self, creds: GmailCredentials, user_id: str = "me") -> Dict[str, Any]:
+        return await self._request(creds, "GET", f"/users/{user_id}/settings/pop")
+
+    async def update_pop(
+        self, creds: GmailCredentials, payload: Dict[str, Any], user_id: str = "me"
+    ) -> Dict[str, Any]:
+        return await self._request(
+            creds, "PUT", f"/users/{user_id}/settings/pop", json_body=payload
+        )
+
+    async def get_language(self, creds: GmailCredentials, user_id: str = "me") -> Dict[str, Any]:
+        return await self._request(creds, "GET", f"/users/{user_id}/settings/language")
+
+    async def update_language(
+        self, creds: GmailCredentials, payload: Dict[str, Any], user_id: str = "me"
+    ) -> Dict[str, Any]:
+        return await self._request(
+            creds, "PUT", f"/users/{user_id}/settings/language", json_body=payload
+        )
+
     async def list_filters(self, creds: GmailCredentials, user_id: str = "me") -> Dict[str, Any]:
         return await self._request(creds, "GET", f"/users/{user_id}/settings/filters")
 
@@ -870,6 +984,13 @@ class GmailService:
             creds, "PATCH", f"/users/{user_id}/settings/sendAs/{send_as_email}", json_body=payload
         )
 
+    async def verify_send_as(
+        self, creds: GmailCredentials, send_as_email: str, user_id: str = "me"
+    ) -> None:
+        await self._request(
+            creds, "POST", f"/users/{user_id}/settings/sendAs/{send_as_email}/verify"
+        )
+
     async def delete_send_as(
         self, creds: GmailCredentials, send_as_email: str, user_id: str = "me"
     ) -> None: