from __future__ import annotations import asyncio import random import time from dataclasses import dataclass, field from typing import Any, Dict, List, Optional import httpx from app.config import get_settings from app.core.logger import get_logger from app.models.schemas import ( GmailComposeMessageRequest, GmailFilterCreateRequest, GmailLabelRequest, GmailParsedMessage, ) from app.utils.gmail_utils import ( GmailMessageValidationError, build_mime_message, parse_message, ) from app.utils.http_utils import SharedAsyncClient _logger = get_logger(__name__) _settings = get_settings() _RETRYABLE_STATUS = frozenset({ int(code) for code in _settings.gmail_retryable_statuses.split(",") if code.strip().isdigit() }) class GmailAPIError(Exception): """Raised for upstream Gmail API failures that map to a client-facing error.""" def __init__(self, message: str, status_code: int = 400, *, reason: str = "") -> None: super().__init__(message) self.message = message self.status_code = status_code self.reason = reason @dataclass class GmailCredentials: """Client-supplied OAuth credentials (never persisted by this backend). ``refreshed_access_token`` is populated whenever the service transparently refreshes the access token, so the stateless client can persist it itself. """ access_token: str refresh_token: Optional[str] = None client_id: Optional[str] = None client_secret: Optional[str] = None expires_at: Optional[float] = None refreshed_access_token: Optional[str] = field(default=None, repr=False) @property def can_refresh(self) -> bool: return bool(self.refresh_token and self.client_id and self.client_secret) @property def needs_proactive_refresh(self) -> bool: if self.expires_at is None: return False return time.time() >= self.expires_at - _settings.gmail_refresh_buffer_seconds class GmailService: """Stateless, fully-async abstraction over the Gmail REST API. Credentials are supplied by the caller on every invocation. Access tokens are transparently refreshed against Google's token endpoint when a ``refresh_token`` + ``client_id`` + ``client_secret`` are provided, and the resulting token is surfaced via ``GmailCredentials.refreshed_access_token``. """ def __init__(self) -> None: self._http = SharedAsyncClient( timeout=httpx.Timeout(_settings.gmail_timeout), follow_redirects=False, ) async def _get_client(self) -> httpx.AsyncClient: return await self._http.get() async def close(self) -> None: await self._http.close() # ------------------------------------------------------------------ # Core request pipeline: auth, retry, backoff, error mapping # ------------------------------------------------------------------ async def _request( self, creds: GmailCredentials, method: str, path: str, *, params: Optional[Dict[str, Any]] = None, json_body: Optional[Any] = None, ) -> Any: """Send a Gmail API request with proactive + on-401 refresh and retries.""" if creds.needs_proactive_refresh: await self._refresh(creds, context=f"{method} {path}") refresh_done = False attempt = 0 while True: response = await self._send( creds, method, path, params=params, json_body=json_body ) if 200 <= response.status_code < 300: return self._decode(response) if response.status_code == 401 and creds.can_refresh and not refresh_done: _logger.warning( "Gmail 401 for %s %s; refreshing access token once", method, path, ) await self._refresh(creds, context=f"{method} {path}") refresh_done = True continue if response.status_code in _RETRYABLE_STATUS and attempt < _settings.gmail_max_retries: await self._backoff(response, attempt) attempt += 1 continue raise self._map_error(response, creds) async def _send( self, creds: GmailCredentials, method: str, path: str, *, params: Optional[Dict[str, Any]] = None, json_body: Optional[Any] = None, ) -> httpx.Response: client = await self._get_client() url = f"{_settings.gmail_api_base_url}{path}" headers = { "Authorization": f"Bearer {creds.access_token}", "Accept": "application/json", } try: return await client.request( method, url, params=params, json=json_body, headers=headers, ) except httpx.TimeoutException as exc: raise GmailAPIError( f"Gmail API request timed out for {method} {path}.", status_code=504 ) from exc except httpx.RequestError as exc: raise GmailAPIError( f"Failed to reach Gmail API: {exc}", status_code=502 ) from exc @staticmethod def _decode(response: httpx.Response) -> Any: if not response.content: return None try: return response.json() except ValueError: return response.text async def _backoff(self, response: httpx.Response, attempt: int) -> None: retry_after = response.headers.get("Retry-After") delay: Optional[float] = None if retry_after: try: delay = float(retry_after) except (TypeError, ValueError): delay = None if delay is None: base = _settings.gmail_base_backoff_seconds * (2 ** attempt) delay = min(base, _settings.gmail_max_backoff_seconds) * (0.5 + random.random()) delay = min(delay, _settings.gmail_max_backoff_seconds + 5.0) _logger.warning( "Gmail transient HTTP %s; retrying in %.2fs (attempt %d/%d)", response.status_code, delay, attempt + 1, _settings.gmail_max_retries, ) await asyncio.sleep(delay) def _map_error(self, response: httpx.Response, creds: GmailCredentials) -> GmailAPIError: status = response.status_code reason = "" detail = "" try: body = response.json() error = body.get("error") or {} if isinstance(error, dict): detail = error.get("message", "") errors = error.get("errors") or [] if errors and isinstance(errors[0], dict): reason = errors[0].get("reason", "") except Exception: body = {} detail = response.text[:500] _logger.error( "Gmail API error: HTTP %s reason=%s detail=%s", status, reason, detail ) if status == 401: if creds.can_refresh: return GmailAPIError( "Access token is invalid or expired and could not be refreshed. " "Re-authorize via POST /api/v1/google/oauth/auth-url.", status_code=401, reason=reason, ) return GmailAPIError( "Access token is invalid or expired. Refresh it via " "POST /api/v1/google/oauth/refresh (or /api/v1/gmail/token/refresh) " "and retry, or supply a fresh access token.", status_code=401, reason=reason, ) if status == 403 and reason == "PERMISSION_DENIED": return GmailAPIError( "Permission denied. The access token may be missing the required " "Gmail scope (see /api/v1/google/gmail/scopes).", status_code=403, reason=reason, ) if status == 429: return GmailAPIError( "Gmail API rate limit exceeded. Please retry after a short delay.", status_code=429, reason=reason, ) if status in _RETRYABLE_STATUS: return GmailAPIError( detail or f"Gmail API service error (HTTP {status}).", status_code=status, reason=reason, ) return GmailAPIError( detail or f"Gmail API error (HTTP {status}).", status_code=status, reason=reason, ) # ------------------------------------------------------------------ # Token lifecycle # ------------------------------------------------------------------ async def refresh_access_token( self, *, client_id: str, client_secret: str, refresh_token: str, ) -> Dict[str, Any]: """Refresh an access token without touching the mailbox (standalone).""" tokens = await self._refresh( creds=None, client_id=client_id, client_secret=client_secret, refresh_token=refresh_token, context="standalone refresh", ) return tokens async def _refresh( self, creds: Optional[GmailCredentials], *, context: str, client_id: Optional[str] = None, client_secret: Optional[str] = None, refresh_token: Optional[str] = None, ) -> Dict[str, Any]: if creds is not None: client_id = creds.client_id client_secret = creds.client_secret refresh_token = creds.refresh_token if not (client_id and client_secret and refresh_token): raise GmailAPIError( "Token refresh requires refresh_token, client_id and client_secret.", status_code=400, ) client = await self._get_client() data = { "client_id": client_id, "client_secret": client_secret, "refresh_token": refresh_token, "grant_type": "refresh_token", } try: response = await client.post(_settings.google_oauth_token_url, data=data) except httpx.TimeoutException as exc: raise GmailAPIError( "Google token refresh timed out.", status_code=504 ) from exc except httpx.RequestError as exc: raise GmailAPIError( f"Failed to reach Google token endpoint: {exc}", status_code=502 ) from exc if response.status_code == 200: tokens = response.json() new_token = tokens.get("access_token") if not new_token: raise GmailAPIError( "Google did not return an access token.", status_code=502 ) expires_in = int(tokens.get("expires_in", 3600)) if creds is not None: creds.access_token = new_token creds.expires_at = time.time() + expires_in creds.refreshed_access_token = new_token _logger.info("Gmail access token refreshed (%s)", context) return tokens try: body = response.json() error = body.get("error", "") description = body.get("error_description", "") except Exception: error = "" description = "" body = {} if error == "invalid_grant": raise GmailAPIError( description or "Refresh token is invalid, expired, or revoked. " "Re-authorize via POST /api/v1/google/oauth/auth-url.", status_code=401, reason=error, ) if error == "invalid_client": raise GmailAPIError( description or "Invalid client_id or client_secret.", status_code=401, reason=error, ) raise GmailAPIError( description or f"Google token refresh failed (HTTP {response.status_code}).", status_code=response.status_code if response.status_code >= 400 else 502, reason=error, ) # ------------------------------------------------------------------ # Users profile & scopes # ------------------------------------------------------------------ async def get_profile(self, creds: GmailCredentials, user_id: str = "me") -> Dict[str, Any]: return await self._request(creds, "GET", f"/users/{user_id}/profile") async def list_available_scopes(self) -> List[Dict[str, Any]]: from app.models.schemas import GmailScope return [ { "name": scope.name, "uri": scope.value, "permission_level": scope.permission_level, } for scope in GmailScope ] # ------------------------------------------------------------------ # Messages # ------------------------------------------------------------------ async def list_messages( self, creds: GmailCredentials, user_id: str = "me", *, q: Optional[str] = None, label_ids: Optional[List[str]] = None, max_results: Optional[int] = None, page_token: Optional[str] = None, include_spam_trash: bool = False, batch_size: Optional[int] = None, ) -> Dict[str, Any]: params: Dict[str, Any] = {} if q is not None: params["q"] = q if label_ids: params["labelIds"] = label_ids if max_results is not None: params["maxResults"] = max_results if page_token: params["pageToken"] = page_token if include_spam_trash: params["includeSpamTrash"] = "true" if batch_size is not None: params["batchSize"] = batch_size return await self._request( creds, "GET", f"/users/{user_id}/messages", params=params or None ) async def get_message( self, creds: GmailCredentials, message_id: str, user_id: str = "me", *, format: str = "full", metadata_headers: Optional[List[str]] = None, headers: Optional[List[str]] = None, ) -> Dict[str, Any]: params: Dict[str, Any] = {"format": format} if metadata_headers: params["metadataHeaders"] = metadata_headers if headers: params["headers"] = headers return await self._request( creds, "GET", f"/users/{user_id}/messages/{message_id}", params=params ) async def get_parsed_message( self, creds: GmailCredentials, message_id: str, user_id: str = "me", *, format: str = "full", ) -> GmailParsedMessage: data = await self.get_message(creds, message_id, user_id=user_id, format=format) return parse_message(data) async def send_raw( self, creds: GmailCredentials, raw: str, user_id: str = "me", *, thread_id: Optional[str] = None, labels: Optional[List[str]] = None, ) -> Dict[str, Any]: body: Dict[str, Any] = {"raw": raw} if thread_id: body["threadId"] = thread_id if labels: body["labelIds"] = labels return await self._request( creds, "POST", f"/users/{user_id}/messages/send", json_body=body ) async def send_composed( self, creds: GmailCredentials, request: GmailComposeMessageRequest, user_id: str = "me", ) -> Dict[str, Any]: try: raw = build_mime_message(request) except GmailMessageValidationError as exc: raise GmailAPIError(str(exc), status_code=400) from exc return await self.send_raw( creds, raw, user_id=user_id, thread_id=request.thread_id, labels=request.labels, ) async def insert_message( self, creds: GmailCredentials, raw: str, user_id: str = "me", *, label_ids: Optional[List[str]] = None, internal_date_source: Optional[str] = None, ) -> 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 return await self._request( 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, message_id: str, user_id: str = "me", *, add_label_ids: Optional[List[str]] = None, remove_label_ids: Optional[List[str]] = None, ) -> Dict[str, Any]: body: Dict[str, Any] = {} if add_label_ids: body["addLabelIds"] = add_label_ids if remove_label_ids: body["removeLabelIds"] = remove_label_ids return await self._request( creds, "POST", f"/users/{user_id}/messages/{message_id}/modify", json_body=body ) async def batch_modify( self, creds: GmailCredentials, ids: List[str], user_id: str = "me", *, add_label_ids: Optional[List[str]] = None, remove_label_ids: Optional[List[str]] = None, ) -> None: body: Dict[str, Any] = {"ids": ids} if add_label_ids: body["addLabelIds"] = add_label_ids if remove_label_ids: body["removeLabelIds"] = remove_label_ids await self._request(creds, "POST", f"/users/{user_id}/messages/batchModify", json_body=body) async def batch_delete( self, creds: GmailCredentials, ids: List[str], user_id: str = "me" ) -> None: await self._request( creds, "POST", f"/users/{user_id}/messages/batchDelete", json_body={"ids": ids} ) async def trash_message( self, creds: GmailCredentials, message_id: str, user_id: str = "me" ) -> Dict[str, Any]: return await self._request( creds, "POST", f"/users/{user_id}/messages/{message_id}/trash" ) async def untrash_message( self, creds: GmailCredentials, message_id: str, user_id: str = "me" ) -> Dict[str, Any]: return await self._request( creds, "POST", f"/users/{user_id}/messages/{message_id}/untrash" ) async def delete_message( self, creds: GmailCredentials, message_id: str, user_id: str = "me" ) -> None: await self._request(creds, "DELETE", f"/users/{user_id}/messages/{message_id}") async def get_attachment( self, creds: GmailCredentials, message_id: str, attachment_id: str, user_id: str = "me", ) -> Dict[str, Any]: return await self._request( creds, "GET", 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 # ------------------------------------------------------------------ async def list_drafts( self, creds: GmailCredentials, user_id: str = "me", *, max_results: Optional[int] = None, page_token: Optional[str] = None, q: Optional[str] = None, include_spam_trash: bool = False, ) -> Dict[str, Any]: params: Dict[str, Any] = {} if max_results is not None: params["maxResults"] = max_results if page_token: params["pageToken"] = page_token if q: params["q"] = q if include_spam_trash: params["includeSpamTrash"] = "true" return await self._request(creds, "GET", f"/users/{user_id}/drafts", params=params or None) async def get_draft( self, creds: GmailCredentials, draft_id: str, user_id: str = "me", *, format: str = "full", ) -> Dict[str, Any]: return await self._request( creds, "GET", f"/users/{user_id}/drafts/{draft_id}", params={"format": format} ) async def create_draft( self, creds: GmailCredentials, raw: str, user_id: str = "me", *, thread_id: Optional[str] = None, ) -> Dict[str, Any]: body: Dict[str, Any] = {"message": {"raw": raw}} if thread_id: body["message"]["threadId"] = thread_id return await self._request(creds, "POST", f"/users/{user_id}/drafts", json_body=body) async def update_draft( self, creds: GmailCredentials, draft_id: str, raw: str, user_id: str = "me", *, thread_id: Optional[str] = None, ) -> Dict[str, Any]: body: Dict[str, Any] = {"message": {"raw": raw}} if thread_id: body["message"]["threadId"] = thread_id return await self._request( creds, "PUT", f"/users/{user_id}/drafts/{draft_id}", json_body=body ) async def send_draft( self, creds: GmailCredentials, draft_id: str, user_id: str = "me" ) -> Dict[str, Any]: return await self._request( creds, "POST", f"/users/{user_id}/drafts/send", json_body={"id": draft_id} ) async def delete_draft( self, creds: GmailCredentials, draft_id: str, user_id: str = "me" ) -> None: await self._request(creds, "DELETE", f"/users/{user_id}/drafts/{draft_id}") # ------------------------------------------------------------------ # Threads # ------------------------------------------------------------------ async def list_threads( self, creds: GmailCredentials, user_id: str = "me", *, q: Optional[str] = None, label_ids: Optional[List[str]] = None, max_results: Optional[int] = None, page_token: Optional[str] = None, include_spam_trash: bool = False, ) -> Dict[str, Any]: params: Dict[str, Any] = {} if q: params["q"] = q if label_ids: params["labelIds"] = label_ids if max_results is not None: params["maxResults"] = max_results if page_token: params["pageToken"] = page_token if include_spam_trash: params["includeSpamTrash"] = "true" return await self._request(creds, "GET", f"/users/{user_id}/threads", params=params or None) async def get_thread( self, creds: GmailCredentials, thread_id: str, user_id: str = "me", *, format: str = "full", metadata_headers: Optional[List[str]] = None, ) -> Dict[str, Any]: params: Dict[str, Any] = {"format": format} if metadata_headers: params["metadataHeaders"] = metadata_headers return await self._request( creds, "GET", f"/users/{user_id}/threads/{thread_id}", params=params ) async def modify_thread( self, creds: GmailCredentials, thread_id: str, user_id: str = "me", *, add_label_ids: Optional[List[str]] = None, remove_label_ids: Optional[List[str]] = None, ) -> Dict[str, Any]: body: Dict[str, Any] = {} if add_label_ids: body["addLabelIds"] = add_label_ids if remove_label_ids: body["removeLabelIds"] = remove_label_ids return await self._request( creds, "POST", f"/users/{user_id}/threads/{thread_id}/modify", json_body=body ) async def trash_thread( self, creds: GmailCredentials, thread_id: str, user_id: str = "me" ) -> Dict[str, Any]: return await self._request( creds, "POST", f"/users/{user_id}/threads/{thread_id}/trash" ) async def untrash_thread( self, creds: GmailCredentials, thread_id: str, user_id: str = "me" ) -> Dict[str, Any]: return await self._request( creds, "POST", f"/users/{user_id}/threads/{thread_id}/untrash" ) async def delete_thread( self, creds: GmailCredentials, thread_id: str, user_id: str = "me" ) -> None: await self._request(creds, "DELETE", f"/users/{user_id}/threads/{thread_id}") # ------------------------------------------------------------------ # Labels # ------------------------------------------------------------------ async def list_labels(self, creds: GmailCredentials, user_id: str = "me") -> Dict[str, Any]: return await self._request(creds, "GET", f"/users/{user_id}/labels") async def get_label( self, creds: GmailCredentials, label_id: str, user_id: str = "me" ) -> Dict[str, Any]: return await self._request(creds, "GET", f"/users/{user_id}/labels/{label_id}") async def create_label( self, creds: GmailCredentials, label: GmailLabelRequest, user_id: str = "me" ) -> Dict[str, Any]: body: Dict[str, Any] = {"name": label.name} if label.label_list_visibility: body["labelListVisibility"] = label.label_list_visibility if label.message_list_visibility: body["messageListVisibility"] = label.message_list_visibility if label.color: body["color"] = label.color return await self._request(creds, "POST", f"/users/{user_id}/labels", json_body=body) async def update_label( self, creds: GmailCredentials, label_id: str, label: GmailLabelRequest, user_id: str = "me", ) -> Dict[str, Any]: body: Dict[str, Any] = {"name": label.name} if label.label_list_visibility: body["labelListVisibility"] = label.label_list_visibility if label.message_list_visibility: body["messageListVisibility"] = label.message_list_visibility if label.color: body["color"] = label.color return await self._request( 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: await self._request(creds, "DELETE", f"/users/{user_id}/labels/{label_id}") # ------------------------------------------------------------------ # Settings: auto-forwarding / vacation / filters / forwarding / send-as / delegates # ------------------------------------------------------------------ async def get_auto_forwarding(self, creds: GmailCredentials, user_id: str = "me") -> Dict[str, Any]: return await self._request(creds, "GET", f"/users/{user_id}/settings/autoForwarding") async def update_auto_forwarding( 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/autoForwarding", json_body=payload ) async def get_vacation(self, creds: GmailCredentials, user_id: str = "me") -> Dict[str, Any]: return await self._request(creds, "GET", f"/users/{user_id}/settings/vacation") async def update_vacation( 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/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") async def get_filter( self, creds: GmailCredentials, filter_id: str, user_id: str = "me" ) -> Dict[str, Any]: return await self._request(creds, "GET", f"/users/{user_id}/settings/filters/{filter_id}") async def create_filter( self, creds: GmailCredentials, req: GmailFilterCreateRequest, user_id: str = "me" ) -> Dict[str, Any]: body: Dict[str, Any] = {"criteria": req.criteria or {}, "action": req.action or {}} return await self._request(creds, "POST", f"/users/{user_id}/settings/filters", json_body=body) async def delete_filter( self, creds: GmailCredentials, filter_id: str, user_id: str = "me" ) -> None: await self._request(creds, "DELETE", f"/users/{user_id}/settings/filters/{filter_id}") async def list_forwarding_addresses(self, creds: GmailCredentials, user_id: str = "me") -> Dict[str, Any]: return await self._request(creds, "GET", f"/users/{user_id}/settings/forwardingAddresses") async def get_forwarding_address( self, creds: GmailCredentials, forwarding_email: str, user_id: str = "me" ) -> Dict[str, Any]: return await self._request( creds, "GET", f"/users/{user_id}/settings/forwardingAddresses/{forwarding_email}" ) async def create_forwarding_address( self, creds: GmailCredentials, forwarding_email: str, user_id: str = "me" ) -> Dict[str, Any]: return await self._request( creds, "POST", f"/users/{user_id}/settings/forwardingAddresses", json_body={"forwardingEmail": forwarding_email}, ) async def delete_forwarding_address( self, creds: GmailCredentials, forwarding_email: str, user_id: str = "me" ) -> None: await self._request( creds, "DELETE", f"/users/{user_id}/settings/forwardingAddresses/{forwarding_email}" ) async def list_send_as(self, creds: GmailCredentials, user_id: str = "me") -> Dict[str, Any]: return await self._request(creds, "GET", f"/users/{user_id}/settings/sendAs") async def get_send_as( self, creds: GmailCredentials, send_as_email: str, user_id: str = "me" ) -> Dict[str, Any]: return await self._request( creds, "GET", f"/users/{user_id}/settings/sendAs/{send_as_email}" ) async def create_send_as( self, creds: GmailCredentials, payload: Dict[str, Any], user_id: str = "me" ) -> Dict[str, Any]: return await self._request( creds, "POST", f"/users/{user_id}/settings/sendAs", json_body=payload ) async def update_send_as( self, creds: GmailCredentials, send_as_email: str, payload: Dict[str, Any], user_id: str = "me", ) -> Dict[str, Any]: return await self._request( 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: await self._request( creds, "DELETE", f"/users/{user_id}/settings/sendAs/{send_as_email}" ) async def list_delegates(self, creds: GmailCredentials, user_id: str = "me") -> Dict[str, Any]: return await self._request(creds, "GET", f"/users/{user_id}/settings/delegates") async def get_delegate( self, creds: GmailCredentials, delegate_email: str, user_id: str = "me" ) -> Dict[str, Any]: return await self._request( creds, "GET", f"/users/{user_id}/settings/delegates/{delegate_email}" ) async def create_delegate( self, creds: GmailCredentials, delegate_email: str, user_id: str = "me" ) -> Dict[str, Any]: return await self._request( creds, "POST", f"/users/{user_id}/settings/delegates", json_body={"delegateEmail": delegate_email}, ) async def delete_delegate( self, creds: GmailCredentials, delegate_email: str, user_id: str = "me" ) -> None: await self._request( creds, "DELETE", f"/users/{user_id}/settings/delegates/{delegate_email}" ) async def list_smime_info( self, creds: GmailCredentials, send_as_email: str, user_id: str = "me" ) -> Dict[str, Any]: return await self._request( creds, "GET", f"/users/{user_id}/settings/sendAs/{send_as_email}/smimeInfo" ) async def get_smime_info( self, creds: GmailCredentials, send_as_email: str, smime_info_id: str, user_id: str = "me", ) -> Dict[str, Any]: return await self._request( creds, "GET", f"/users/{user_id}/settings/sendAs/{send_as_email}/smimeInfo/{smime_info_id}", ) async def insert_smime_info( self, creds: GmailCredentials, send_as_email: str, payload: Dict[str, Any], user_id: str = "me", ) -> Dict[str, Any]: return await self._request( creds, "POST", f"/users/{user_id}/settings/sendAs/{send_as_email}/smimeInfo", json_body=payload, ) async def delete_smime_info( self, creds: GmailCredentials, send_as_email: str, smime_info_id: str, user_id: str = "me", ) -> None: await self._request( creds, "DELETE", f"/users/{user_id}/settings/sendAs/{send_as_email}/smimeInfo/{smime_info_id}", ) async def set_default_smime_info( self, creds: GmailCredentials, send_as_email: str, smime_info_id: str, user_id: str = "me", ) -> None: await self._request( creds, "POST", f"/users/{user_id}/settings/sendAs/{send_as_email}/smimeInfo/{smime_info_id}/setDefault", )