Gmail Search

Inputs

Outputs

Source

python

# Input preparation
inp = __ivy_ctx__["nodes"][__ivy_node_id__]["input"]

# Compute
# ivy_connect, embedded from IVYX Flows' connectors/sdk by tools/author-connector-nodes.py; never edited here.
import sys as _sys, types as _types
_IVY_CONNECT = {'errors': '"""The errors a connector node raises, one per thing a person would do about it."""\n\nfrom typing import Any, Optional\n\n\nclass ConnectorError(Exception):\n    """An app answered with an error, or could not be reached. ``status`` is 0 when nothing answered."""\n\n    def __init__(self, message: str, status: int = 0, body: Any = None):\n        super().__init__(message)\n        self.status = status\n        self.body = body\n\n\nclass AuthError(ConnectorError):\n    """The app refused the credential (401/403): the connection needs a look, or a new sign-in."""\n\n\nclass NotFound(ConnectorError):\n    """404: what the step asked for is not there."""\n\n\nclass RateLimited(ConnectorError):\n    """429 after every retry: the app wants fewer calls."""\n\n\nclass Invalid(ConnectorError):\n    """400, 409 or 422: the app refused what the step sent, and says why."""\n\n\nclass Unavailable(ConnectorError):\n    """5xx after every retry, a timeout, or no answer at all."""\n\n\ndef error_for(status: int, message: str, body: Any = None) -> ConnectorError:\n    if status in (401, 403):\n        return AuthError(message, status, body)\n    if status == 404:\n        return NotFound(message, status, body)\n    if status == 429:\n        return RateLimited(message, status, body)\n    if status in (400, 409, 410, 412, 415, 422):\n        return Invalid(message, status, body)\n    if status == 0 or status >= 500:\n        return Unavailable(message, status, body)\n    return ConnectorError(message, status, body)\n\n\ndef app_message(body: Any) -> Optional[str]:\n    """The app\'s own words for an error, from the shapes APIs answer with."""\n    if not isinstance(body, dict):\n        return None\n    err = body.get("error")\n    if isinstance(err, dict) and isinstance(err.get("message"), str):\n        return err["message"][:300]\n    for key in ("error_description", "message", "error", "detail"):\n        if isinstance(body.get(key), str) and body[key]:\n            return body[key][:300]\n    errors = body.get("errors")\n    if isinstance(errors, list) and errors and isinstance(errors[0], dict) and isinstance(errors[0].get("message"), str):\n        return errors[0]["message"][:300]\n    return None\n', 'client': '"""The client a connector node calls its app with."""\n\nimport email.utils\nimport json as jsonlib\nimport os\nimport random\nimport threading\nimport time\nimport urllib.error\nimport urllib.parse\nimport urllib.request\nfrom typing import Any, Callable, Dict, Iterator, List, Optional, Union\n\nfrom .errors import AuthError, ConnectorError, Unavailable, app_message, error_for\n\nIDEMPOTENT = ("GET", "HEAD", "OPTIONS", "PUT", "DELETE")\nRETRY_STATUS = (500, 502, 503, 504)\nMAX_WAIT = 60.0\n\n\nclass Response:\n    """What an app answered: the status, the headers (lower-case names), and the body as text, bytes or JSON."""\n\n    def __init__(self, status: int, headers: Dict[str, str], content: bytes, url: str):\n        self.status = status\n        self.headers = headers\n        self.content = content\n        self.url = url\n\n    @property\n    def ok(self) -> bool:\n        return self.status < 400\n\n    @property\n    def text(self) -> str:\n        return self.content.decode("utf-8", errors="replace")\n\n    def json(self) -> Any:\n        if not self.content.strip():\n            return None\n        return jsonlib.loads(self.text)\n\n    def maybe_json(self) -> Any:\n        try:\n            return self.json()\n        except ValueError:\n            return None\n\n\nclass _Bucket:\n    """At most ``per_minute`` calls a minute per connection, shared by every client of it in this process."""\n\n    def __init__(self, per_minute: float):\n        self.rate = per_minute / 60.0\n        self.capacity = max(1.0, min(per_minute, 10.0))\n        self.tokens = self.capacity\n        self.at = time.monotonic()\n        self.lock = threading.Lock()\n\n    def take(self, sleep: Callable[[float], None]) -> None:\n        with self.lock:\n            now = time.monotonic()\n            self.tokens = min(self.capacity, self.tokens + (now - self.at) * self.rate)\n            self.at = now\n            wait = 0.0 if self.tokens >= 1 else (1 - self.tokens) / self.rate\n            self.tokens -= 1\n        if wait > 0:\n            sleep(wait)\n\n\n_buckets: Dict[str, _Bucket] = {}\n\n\ndef _dig(value: Any, path: str) -> Any:\n    for key in path.split("."):\n        if isinstance(value, dict):\n            value = value.get(key)\n        elif isinstance(value, list) and key.isdigit() and int(key) < len(value):\n            value = value[int(key)]\n        else:\n            return None\n    return value\n\n\ndef _next_link(header: str) -> Optional[str]:\n    """The ``rel="next"`` URL of a Link header, as GitHub and others page with."""\n    for part in header.split(","):\n        bits = part.split(";")\n        if len(bits) > 1 and any(b.strip().replace(" ", "") in (\'rel="next"\', "rel=next") for b in bits[1:]):\n            return bits[0].strip().strip("<>")\n    return None\n\n\ndef _retry_after(value: Optional[str]) -> Optional[float]:\n    if not value:\n        return None\n    try:\n        return max(0.0, float(value))\n    except ValueError:\n        pass\n    try:\n        when = email.utils.parsedate_to_datetime(value)\n        return max(0.0, when.timestamp() - time.time())\n    except (TypeError, ValueError):\n        return None\n\n\nclass Client:\n    """\n    A connection\'s client. Every request carries the connection\'s headers and\n    query; a path is under its base URL, a full URL is used as it is.\n    """\n\n    def __init__(\n        self,\n        credential: Dict[str, Any],\n        timeout: float = 30.0,\n        retries: int = 4,\n        broker_url: Optional[str] = None,\n        broker_key: Optional[str] = None,\n        sleep: Callable[[float], None] = time.sleep,\n        base: Optional[str] = None,\n    ):\n        if not isinstance(credential, dict) or not isinstance(credential.get("headers"), dict):\n            raise ConnectorError(\'The connection input is not a connection; give the step {"$secret": "connection.<id>"}.\')\n        self.credential = credential\n        # One of the connector\'s other roots (``bases``), for an app spread over hosts: Google\'s gmail. and sheets.\n        self.base = base\n        if base is not None and base not in (credential.get("bases") or {}):\n            raise ConnectorError(f"The connection {credential.get(\'id\')} has no base called {base}; its connector names {\', \'.join(sorted(credential.get(\'bases\') or {})) or \'none\'}.")\n        self.timeout = timeout\n        self.retries = retries\n        self.sleep = sleep\n        self.broker_url = broker_url if broker_url is not None else os.environ.get("IVYX_CONNECTIONS_URL", "")\n        self.broker_key = broker_key if broker_key is not None else os.environ.get("IVYX_CONNECTIONS_KEY", "")\n        per_minute = (credential.get("rateLimit") or {}).get("perMinute")\n        self._bucket = None\n        if per_minute:\n            self._bucket = _buckets.setdefault(str(credential.get("id")), _Bucket(float(per_minute)))\n\n    # ---------- who and where ----------\n\n    @property\n    def name(self) -> str:\n        return str(self.credential.get("connector") or self.credential.get("id") or "the app")\n\n    @property\n    def base_url(self) -> str:\n        if self.base is not None:\n            return str((self.credential.get("bases") or {}).get(self.base) or "")\n        return str(self.credential.get("baseUrl") or "")\n\n    @property\n    def account(self) -> str:\n        """Who the connection signs in as, from its test: an address, a workspace."""\n        return str(self.credential.get("account") or "")\n\n    def url(self, path: str) -> str:\n        if path.lower().startswith(("http://", "https://")):\n            return path\n        if not self.base_url:\n            raise ConnectorError(f"{path} is relative and the connection {self.credential.get(\'id\')} has no base URL; give a full URL.")\n        return self.base_url.rstrip("/") + "/" + path.lstrip("/")\n\n    # ---------- tokens ----------\n\n    def _secrets(self) -> List[str]:\n        """Every value that signs in, so no message ever carries one."""\n        out = [str(v) for v in (self.credential.get("values") or {}).values() if v]\n        out.extend(str(t.get("access_token")) for t in (self.credential.get("tokens") or {}).values() if isinstance(t, dict) and t.get("access_token"))\n        for v in (self.credential.get("headers") or {}).values():\n            out.extend(p for p in str(v).split(" ") if len(p) > 8)\n        return [s for s in out if len(s) >= 6]\n\n    def redact(self, text: str) -> str:\n        for s in self._secrets():\n            text = text.replace(s, "[REDACTED]")\n        return text\n\n    def refresh(self) -> bool:\n        """Ask the broker for a fresh credential; False when there is no broker to ask."""\n        if not self.broker_url or not self.broker_key or not self.credential.get("id"):\n            return False\n        req = urllib.request.Request(\n            f"{self.broker_url.rstrip(\'/\')}/{urllib.parse.quote(str(self.credential[\'id\']))}/credentials?fresh=1",\n            headers={"Authorization": f"Bearer {self.broker_key}", "Accept": "application/json"},\n        )\n        try:\n            with urllib.request.urlopen(req, timeout=30) as res:\n                fresh = jsonlib.loads(res.read().decode("utf-8"))\n        except urllib.error.HTTPError as err:\n            with err:\n                said = app_message(_loads(err.read())) or f"the broker answered {err.code}"\n            raise AuthError(f"The connection {self.credential.get(\'id\')} could not be refreshed: {said}", err.code)\n        except (urllib.error.URLError, OSError, ValueError) as err:\n            raise Unavailable(f"The connections broker did not answer: {err}")\n        if isinstance(fresh, dict) and isinstance(fresh.get("headers"), dict):\n            self.credential = fresh\n            return True\n        return False\n\n    def token(self, audience: Optional[str] = None) -> str:\n        """\n        The access token, for a protocol that is not HTTP (IMAP\'s XOAUTH2).\n        ``audience`` names a token for another resource the connector asks\n        for (Microsoft\'s ``imap``), handed out beside the main one.\n        """\n        self._ahead()\n        if audience is None:\n            values = self.credential.get("values") or {}\n            found = values.get("access_token") or values.get("token")\n            if not found:\n                raise AuthError(f"The connection {self.credential.get(\'id\')} signs in with {self.credential.get(\'type\')}, not a token.")\n            return str(found)\n        got = (self.credential.get("tokens") or {}).get(audience)\n        if not isinstance(got, dict) or not got.get("access_token"):\n            raise AuthError(f"The connection {self.credential.get(\'id\')} has no token for {audience}: sign in to it again, so the app is asked for that access too.")\n        return str(got["access_token"])\n\n    def _ahead(self) -> None:\n        expires = self.credential.get("expiresAt")\n        if self.credential.get("type") == "oauth2" and isinstance(expires, (int, float)) and expires / 1000.0 - time.time() < 60:\n            self.refresh()\n\n    # ---------- requests ----------\n\n    def request(\n        self,\n        method: str,\n        path: str,\n        params: Optional[Dict[str, Any]] = None,\n        json: Any = None,\n        data: Union[bytes, str, Dict[str, Any], None] = None,\n        headers: Optional[Dict[str, str]] = None,\n        timeout: Optional[float] = None,\n        retry_unsafe: bool = False,\n        check: bool = True,\n    ) -> Response:\n        """\n        One call, tried again on 429 always and on 5xx or no answer when the\n        method is safe to repeat (or ``retry_unsafe``), waiting as the app\n        asks or longer each time. A refused token is refreshed once. With\n        ``check`` an error status raises, else the response comes back as it is.\n        """\n        method = method.upper()\n        self._ahead()\n        repeatable = method in IDEMPOTENT or retry_unsafe\n        refreshed = False\n        attempt = 0\n        while True:\n            if self._bucket:\n                self._bucket.take(self.sleep)\n            try:\n                res = self._send(method, path, params, json, data, headers, timeout)\n            except Unavailable:\n                if repeatable and attempt < self.retries:\n                    self.sleep(self._backoff(attempt))\n                    attempt += 1\n                    continue\n                raise\n            if res.status == 401 and not refreshed and self.credential.get("type") == "oauth2" and self.refresh():\n                refreshed = True\n                continue\n            if attempt < self.retries and (res.status == 429 or (res.status in RETRY_STATUS and repeatable)):\n                wait = _retry_after(res.headers.get("retry-after"))\n                self.sleep(min(MAX_WAIT, wait if wait is not None else self._backoff(attempt)))\n                attempt += 1\n                continue\n            if check and not res.ok:\n                raise self._error(method, path, res)\n            return res\n\n    def _backoff(self, attempt: int) -> float:\n        return min(30.0, 0.5 * (2 ** attempt)) + random.uniform(0, 0.25)\n\n    def _send(self, method, path, params, json, data, headers, timeout) -> Response:\n        url = self.url(path)\n        query = dict(self.credential.get("query") or {})\n        query.update({k: v for k, v in (params or {}).items() if v is not None})\n        if query:\n            sep = "&" if urllib.parse.urlsplit(url).query else "?"\n            url = url + sep + urllib.parse.urlencode(query, doseq=True)\n        sent = {"Accept": "application/json"}\n        sent.update(self.credential.get("headers") or {})\n        body = None\n        if json is not None and data is not None:\n            raise ConnectorError("Give json or data, not both.")\n        if json is not None:\n            body = jsonlib.dumps(json).encode("utf-8")\n            sent["Content-Type"] = "application/json"\n        elif isinstance(data, dict):\n            body = urllib.parse.urlencode(data, doseq=True).encode("utf-8")\n            sent["Content-Type"] = "application/x-www-form-urlencoded"\n        elif isinstance(data, str):\n            body = data.encode("utf-8")\n        elif data is not None:\n            body = data\n        sent.update(headers or {})\n        req = urllib.request.Request(url, data=body, method=method, headers=sent)\n        try:\n            with urllib.request.urlopen(req, timeout=timeout or self.timeout) as res:\n                return Response(res.status, {k.lower(): v for k, v in res.headers.items()}, res.read(), res.geturl())\n        except urllib.error.HTTPError as err:\n            with err:\n                return Response(err.code, {k.lower(): v for k, v in (err.headers or {}).items()}, err.read() or b"", url)\n        except (urllib.error.URLError, OSError) as err:\n            reason = getattr(err, "reason", err)\n            raise Unavailable(self.redact(f"{self.name} could not be reached at {self.url(path)}: {reason}"))\n\n    def _error(self, method: str, path: str, res: Response) -> ConnectorError:\n        body = res.maybe_json()\n        said = app_message(body) or res.text.strip()[:200] or "no reason given"\n        where = path if not path.lower().startswith("http") else urllib.parse.urlsplit(path).path\n        return error_for(res.status, self.redact(f"{res.status} from {self.name} {method} {where}: {said}"), body)\n\n    def get(self, path: str, **kw) -> Response:\n        return self.request("GET", path, **kw)\n\n    def post(self, path: str, **kw) -> Response:\n        return self.request("POST", path, **kw)\n\n    def put(self, path: str, **kw) -> Response:\n        return self.request("PUT", path, **kw)\n\n    def patch(self, path: str, **kw) -> Response:\n        return self.request("PATCH", path, **kw)\n\n    def delete(self, path: str, **kw) -> Response:\n        return self.request("DELETE", path, **kw)\n\n    # ---------- lists ----------\n\n    def paginate(\n        self,\n        method: str,\n        path: str,\n        items: str,\n        style: str = "cursor",\n        next: Optional[str] = None,\n        param: Optional[str] = None,\n        params: Optional[Dict[str, Any]] = None,\n        json: Any = None,\n        page_size: Optional[int] = None,\n        size_param: str = "limit",\n        limit: Optional[int] = None,\n        max_pages: int = 1000,\n    ) -> Iterator[Any]:\n        """\n        Every item of a list the app hands out in pages. ``items`` is the dot\n        path to a page\'s list. ``style`` is how the next page is asked for:\n\n        - ``cursor``: ``next`` is the dot path to the next cursor in a page,\n          sent back as ``param`` (``cursor`` by default; in the JSON body of a\n          POST when there is one). An empty cursor ends it.\n        - ``link``: the ``Link`` header\'s ``rel="next"`` URL.\n        - ``offset``: ``param`` (``offset``) grows by the page\'s length; a page\n          shorter than ``page_size`` ends it.\n        - ``page``: ``param`` (``page``) counts up from 1 until a page is empty.\n        """\n        params = dict(params or {})\n        if page_size:\n            params[size_param] = page_size\n        body = dict(json) if isinstance(json, dict) else json\n        given = 0\n        url = path\n        position: Any = None\n        if style == "offset":\n            param = param or "offset"\n            position = int(params.get(param, 0))\n        elif style == "page":\n            param = param or "page"\n            position = int(params.get(param, 1))\n        elif style == "cursor":\n            param = param or "cursor"\n            if not next:\n                raise ConnectorError("A cursor page names where its next cursor is: next=\'…\'.")\n        elif style != "link":\n            raise ConnectorError(f"A page style is cursor, link, offset or page, not {style}.")\n        for _ in range(max_pages):\n            if style in ("offset", "page"):\n                params[param] = position\n            res = self.request(method, url, params=params, json=body)\n            page = res.json()\n            found = _dig(page, items)\n            if not isinstance(found, list):\n                raise ConnectorError(f"{self.name} answered a page with no list at {items}.")\n            for item in found:\n                yield item\n                given += 1\n                if limit is not None and given >= limit:\n                    return\n            if style == "cursor":\n                cursor = _dig(page, next)\n                if not cursor:\n                    return\n                if isinstance(body, dict) and method.upper() != "GET":\n                    body[param] = cursor\n                else:\n                    params[param] = cursor\n            elif style == "link":\n                link = _next_link(res.headers.get("link", ""))\n                if not link:\n                    return\n                url, params = link, {}\n            elif style == "offset":\n                if not found or (page_size and len(found) < page_size):\n                    return\n                position += len(found)\n            else:\n                if not found:\n                    return\n                position += 1\n\n    # ---------- files ----------\n\n    def download(self, path: str, dest: str, params: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:\n        """A file from the app into the workspace, at ``dest``; folders on the way are made."""\n        res = self.request("GET", path, params=params, headers={"Accept": "*/*"})\n        folder = os.path.dirname(dest)\n        if folder:\n            os.makedirs(folder, exist_ok=True)\n        with open(dest, "wb") as f:\n            f.write(res.content)\n        return {"path": dest, "bytes": len(res.content), "content_type": res.headers.get("content-type", "")}\n\n    def upload(self, path: str, src: str, method: str = "POST", content_type: str = "application/octet-stream", params: Optional[Dict[str, Any]] = None) -> Response:\n        """A workspace file\'s bytes as the body of one request."""\n        with open(src, "rb") as f:\n            content = f.read()\n        return self.request(method, path, params=params, data=content, headers={"Content-Type": content_type})\n\n\ndef _loads(raw: bytes) -> Any:\n    try:\n        return jsonlib.loads(raw.decode("utf-8"))\n    except ValueError:\n        return None\n\n\ndef xoauth2(user: str, token: str) -> bytes:\n    """The SASL XOAUTH2 string IMAP and SMTP sign in with: ``imap.authenticate("XOAUTH2", lambda _: xoauth2(user, token))``."""\n    return f"user={user}\\x01auth=Bearer {token}\\x01\\x01".encode("utf-8")\n\n\ndef client(connection: Union[str, Dict[str, Any]], **kw) -> Client:\n    """The client for a step\'s connection input: the credential\'s JSON text as the service resolved it, or the object."""\n    if isinstance(connection, str):\n        try:\n            connection = jsonlib.loads(connection)\n        except ValueError:\n            raise ConnectorError(\'The connection input is not a connection; give the step {"$secret": "connection.<id>"}.\')\n    return Client(connection, **kw)\n', '__init__': '"""\nivy_connect: what every connector node shares (docs/connectors.md, module 4).\n\nA node gets its connection as a stored secret, ``{"$secret": "connection.<id>"}``,\nwhich the service resolves when the step starts into the credential\'s JSON:\nthe headers and query to send, the base URL, the raw values. This library\nturns that into a client that signs every request, waits and tries again when\nthe app is busy, pages through lists, says errors in the app\'s own words, and\nnever writes a token into a message.\n\n    from ivy_connect import client\n\n    slack = client(inp["connection"])\n    for ch in slack.paginate("GET", "conversations.list", items="channels", next="response_metadata.next_cursor"):\n        ...\n    slack.post("chat.postMessage", json={"channel": inp["channel"], "text": inp["text"]})\n\nAn app spread over hosts names its other roots in the manifest\'s ``bases``;\n``client(inp["connection"], base="sheets")`` calls under one of them. A\nprotocol that is not HTTP takes ``c.token()`` (or ``c.token("imap")`` for a\ntoken of another resource) and ``xoauth2(user, token)``.\n\nThe standard library only, and Python 3.9, like the nodes it runs in.\n"""\n\nfrom .client import Client, Response, client, xoauth2\nfrom .errors import AuthError, ConnectorError, Invalid, NotFound, RateLimited, Unavailable\n\n__all__ = ["client", "xoauth2", "Client", "Response", "ConnectorError", "AuthError", "NotFound", "RateLimited", "Invalid", "Unavailable"]\n__version__ = "0.2.0"\n'}
if getattr(_sys.modules.get('ivy_connect'), '__embedded__', None) != 'd8e4d4dcdcae1a17':
    _pkg = _types.ModuleType('ivy_connect')
    _pkg.__path__ = []
    _pkg.__package__ = 'ivy_connect'
    _sys.modules['ivy_connect'] = _pkg
    for _name in ('errors', 'client'):
        _mod = _types.ModuleType('ivy_connect.' + _name)
        _mod.__package__ = 'ivy_connect'
        _sys.modules['ivy_connect.' + _name] = _mod
        exec(compile(_IVY_CONNECT[_name], '<ivy_connect/' + _name + '.py>', 'exec'), _mod.__dict__)
        setattr(_pkg, _name, _mod)
    exec(compile(_IVY_CONNECT['__init__'], '<ivy_connect/__init__.py>', 'exec'), _pkg.__dict__)
    _pkg.__embedded__ = 'd8e4d4dcdcae1a17'
del _IVY_CONNECT

import base64
import datetime
import email.utils
import os
import re
import urllib.parse
from html.parser import HTMLParser

class _Text(HTMLParser):
    BLOCK = {"p", "div", "br", "li", "tr", "h1", "h2", "h3", "h4", "h5", "h6", "table", "ul", "ol", "blockquote", "pre"}
    def __init__(self):
        super().__init__(); self.parts = []; self.skip = 0
    def handle_starttag(self, tag, attrs):
        if tag in ("script", "style", "head"): self.skip += 1
        if tag in self.BLOCK: self.parts.append("\n")
    def handle_endtag(self, tag):
        if tag in ("script", "style", "head"): self.skip = max(0, self.skip - 1)
        if tag in self.BLOCK: self.parts.append("\n")
    def handle_data(self, data):
        if not self.skip: self.parts.append(data)

def html_to_text(html):
    p = _Text(); p.feed(html); p.close()
    rows = [re.sub(r"[ \t\r\f\v]+", " ", l).strip() for l in "".join(p.parts).split("\n")]
    return "\n".join(l for l in rows if l)

def b64(data):
    return base64.urlsafe_b64decode((data or "") + "=" * (-len(data or "") % 4))

def header(payload, name):
    for h in payload.get("headers") or []:
        if str(h.get("name", "")).lower() == name.lower():
            return str(h.get("value", ""))
    return ""

def walk(part):
    yield part
    for p in part.get("parts") or []:
        yield from walk(p)

def body_text(payload):
    plain = html = None
    for p in walk(payload):
        data = (p.get("body") or {}).get("data")
        if p.get("filename") or not data:
            continue
        if p.get("mimeType") == "text/plain" and plain is None:
            plain = b64(data).decode("utf-8", "replace")
        elif p.get("mimeType") == "text/html" and html is None:
            html = b64(data).decode("utf-8", "replace")
    if plain and plain.strip():
        return plain.strip()
    return html_to_text(html) if html else ""

def when(date, internal_ms):
    try:
        return email.utils.parsedate_to_datetime(date).isoformat()
    except Exception:
        if internal_ms:
            return datetime.datetime.fromtimestamp(int(internal_ms) / 1000, datetime.timezone.utc).isoformat()
        return date

def safe_name(name):
    name = os.path.basename(str(name or "attachment")).strip() or "attachment"
    return re.sub(r"[^A-Za-z0-9._-]+", "_", name)

def read_message(gmail, mid, save, folder):
    """One message as a flow reads it; its attachments written under `folder` when `save`."""
    m = gmail.get("users/me/messages/" + urllib.parse.quote(str(mid), safe=""), params={"format": "full"}).json() or {}
    payload = m.get("payload") or {}
    attachments = []
    for p in walk(payload):
        if not p.get("filename"):
            continue
        body = p.get("body") or {}
        a = {"name": p["filename"], "mime_type": p.get("mimeType", ""), "bytes": int(body.get("size") or 0), "attachment_id": body.get("attachmentId", "")}
        if save:
            data = body.get("data")
            if not data and body.get("attachmentId"):
                data = (gmail.get("users/me/messages/%s/attachments/%s" % (urllib.parse.quote(str(mid), safe=""), urllib.parse.quote(body["attachmentId"], safe=""))).json() or {}).get("data", "")
            raw = b64(data)
            os.makedirs(folder, exist_ok=True)
            path = os.path.join(folder, "%s-%s" % (mid, safe_name(p["filename"])))
            with open(path, "wb") as f:
                f.write(raw)
            a["path"] = path
            a["bytes"] = len(raw)
        attachments.append(a)
    return {
        "id": str(m.get("id", mid)),
        "thread_id": str(m.get("threadId", "")),
        "history_id": str(m.get("historyId", "")),
        "labels": m.get("labelIds") or [],
        "from": header(payload, "From"),
        "to": header(payload, "To"),
        "subject": header(payload, "Subject"),
        "date": when(header(payload, "Date"), m.get("internalDate")),
        "snippet": m.get("snippet", ""),
        "text": body_text(payload),
        "has_attachments": bool(attachments),
        "attachments": attachments,
    }

from ivy_connect import NotFound, client


class GmailSearch:
    "Gmail Search"

    def run(self, inp):
        out = {}
        gmail = client(inp["connection"], base="gmail")
        query = str(inp.get("query") or "").strip()
        labels = [str(l) for l in (inp.get("label_ids") or []) if str(l).strip()]
        limit = int(inp.get("limit") or 20)
        if limit < 1:
            raise ValueError("limit must be at least 1.")
        save = bool(inp.get("save_attachments", False))
        folder = str(inp.get("attachments_dir") or "mail/attachments")
        cursor = str(inp.get("$cursor") or "").strip()
        start = inp.get("$start")

        def search(n):
            ids, token = [], None
            while len(ids) < n:
                page = gmail.get("users/me/messages", params={"q": query or None, "labelIds": labels or None, "maxResults": min(500, n - len(ids)), "pageToken": token}).json() or {}
                ids.extend(str(m["id"]) for m in page.get("messages") or [])
                token = page.get("nextPageToken")
                if not token:
                    break
            return ids[:n]

        def history_now():
            return str((gmail.get("users/me/profile").json() or {}).get("historyId", ""))

        def received(message_labels):
            """New mail, not a draft being written or a message the account sent itself."""
            if "DRAFT" in message_labels or ("SENT" in message_labels and "INBOX" not in message_labels):
                return False
            return all(l in message_labels for l in labels)

        if cursor:
            # What arrived since the cursor, oldest first in the history; a burst
            # larger than the limit is still read whole, up to a hundred.
            added, token, latest = [], None, cursor
            try:
                while True:
                    page = gmail.get("users/me/history", params={"startHistoryId": cursor, "historyTypes": "messageAdded", "labelId": labels[0] if len(labels) == 1 else None, "maxResults": 500, "pageToken": token}).json() or {}
                    for h in page.get("history") or []:
                        for a in h.get("messagesAdded") or []:
                            m = a.get("message") or {}
                            if m.get("id") and str(m["id"]) not in added and received(m.get("labelIds") or []):
                                added.append(str(m["id"]))
                    latest = str(page.get("historyId") or latest)
                    token = page.get("nextPageToken")
                    if not token:
                        break
                if query and added:
                    # The history cannot be searched; what matches is what a search of the newest finds among them.
                    matching = set(search(min(500, max(100, 2 * len(added)))))
                    added = [m for m in added if m in matching]
                ids = list(reversed(added))[: max(limit, 100)]
                next_cursor = latest
            except NotFound:
                # Gmail keeps about a week of history: an older cursor is read again
                # from the newest, and the trigger's keys keep what was started from starting twice.
                next_cursor = history_now()
                ids = search(limit)
        elif start == "now":
            ids, next_cursor = [], history_now()
        else:
            # The history id first, so a message that lands during the search is read next time.
            next_cursor = history_now()
            ids = search(limit)

        messages = [read_message(gmail, i, save, folder) for i in ids]

        out["messages"] = messages
        out["ids"] = [m["id"] for m in messages]
        out["count"] = len(messages)
        out["next_cursor"] = next_cursor
        out["attachment_paths"] = [a["path"] for m in messages for a in m["attachments"] if a.get("path")]
        return out


result = GmailSearch().run(inp)

# Output collection (the runner reads it here)
out = __ivy_ctx__["nodes"][__ivy_node_id__]["output"]
out.update(result)

Tests

Requires: python:3.9

• search

  A Gmail search, newest first: sender, subject, date and text, an HTML body as text, and the history id to resume from.

• attachments

  With save_attachments each attachment is written under the folder and its path given.

• first-read-now

  A trigger's first read under start: now reads no mail and answers where the mailbox stands.

• resume

  From a cursor only the mail that arrived since, newest first, without drafts or sent mail, and the next history id.

• resume-search

  From a cursor with a search, only the new mail the search also finds.

• nothing-new

  Nothing since the cursor is no message and the same history id.

• cursor-too-old

  A history id older than Gmail keeps is read again from the newest, with a fresh history id.

• refused

  A token Google refuses is said as an authentication error, in Google's words.

• not-a-connection

  A connection input that is not a connection's JSON is refused before any call, naming the stored secret to give.