Slack Post Message

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

from ivy_connect import AuthError, ConnectorError, NotFound, RateLimited

# What Slack's error codes mean to the person who has to fix them.
WORDS = {
    "channel_not_found": "Slack has no channel %s that the app can see: give the channel's id (C…), or invite the app to a private channel.",
    "not_in_channel": "The app is not in %s: invite it there (/invite @the-app), or give it the chat:write.public scope for public channels.",
    "is_archived": "%s is archived.",
    "msg_too_long": "The message is longer than Slack takes.",
    "no_text": "The message has no text and no blocks.",
    "missing_scope": "The Slack app lacks a scope this needs (%s); add it under OAuth & Permissions and install the app again.",
}
AUTH = ("invalid_auth", "not_authed", "token_revoked", "token_expired", "account_inactive", "no_permission", "ekm_access_denied")

def checked(answer, about=""):
    """Slack answers most refusals with 200 and ok: false; this raises them in words."""
    if isinstance(answer, dict) and answer.get("ok"):
        return answer
    code = str((answer or {}).get("error") or "unknown_error")
    if code in AUTH:
        raise AuthError("Slack refused the connection (%s); sign in to Slack again." % code, 401, answer)
    if code == "ratelimited":
        raise RateLimited("Slack asks for fewer calls (ratelimited).", 429, answer)
    words = WORDS.get(code)
    if words:
        detail = (answer or {}).get("needed") if code == "missing_scope" else about
        message = words % detail if "%s" in words else words
    else:
        message = "Slack refused it: %s." % code
    raise (NotFound if code == "channel_not_found" else ConnectorError)(message + " (%s)" % code if words else message, 400, answer)

from ivy_connect import client


class SlackPostMessage:
    "Slack Post Message"

    def run(self, inp):
        out = {}
        slack = client(inp["connection"])
        channel = str(inp.get("channel") or "").strip()
        text = str(inp.get("text") or "")
        blocks = inp.get("blocks")
        if not channel:
            raise ValueError("channel is empty; give the channel's id (C…) or name.")
        if not text.strip() and not blocks:
            raise ValueError("text is empty; give the message.")
        body = {"channel": channel, "text": text, "unfurl_links": bool(inp.get("unfurl_links", False))}
        if blocks:
            body["blocks"] = blocks
        if inp.get("thread_ts"):
            body["thread_ts"] = str(inp["thread_ts"])
        answer = checked(slack.post("chat.postMessage", json=body, headers={"Content-Type": "application/json; charset=utf-8"}).json(), channel)

        out["channel"] = str(answer.get("channel", channel))
        out["ts"] = str(answer.get("ts", ""))
        out["thread_ts"] = str((answer.get("message") or {}).get("thread_ts") or "")
        out["text"] = str((answer.get("message") or {}).get("text", text))
        return out


result = SlackPostMessage().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

• post

  A message is posted to the channel and named by its ts.

• thread

  A reply goes in the thread of the message named.

• not-in-channel

  Slack's not_in_channel is said as what to do about it.

• revoked

  A revoked token is an authentication error that asks for a new sign-in.

• no-text

  A message with nothing in it is refused before any call.

• not-a-connection

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