x0hello.py

import os
from slack import RTMClient
from slack.errors import SlackApiError


@RTMClient.run_on(event="message")
async def say_hello(**payload):
    data = payload["data"]
    web_client = payload["web_client"]
    rtm_client = payload["rtm_client"]
    if "text" in data and "Hello" in data.get("text", []):
        channel_id = data["channel"]
        thread_ts = data["ts"]
        user = data["user"]

        try:
            response = await web_client.chat_postMessage(
                channel=channel_id, text=f"Hi <@{user}>!", thread_ts=thread_ts
            )
            print("ok", response)
        except SlackApiError as e:
            # You will get a SlackApiError if "ok" is False
            assert e.response["ok"] is False
            assert e.response["error"]  # str like 'invalid_auth', 'channel_not_found'
            print(f"Got an error: {e.response['error']}")


if __name__ == "__main__":
    import asyncio

    rtm_client = RTMClient(token=os.environ["SLACK_API_TOKEN"], run_async=True)
    asyncio.run(rtm_client.start(), debug=True)

x1hello.py

import os
from slack import RTMClient
from slack.errors import SlackApiError


@RTMClient.run_on(event="message")
async def say_hello(**payload):
    data = payload["data"]
    web_client = payload["web_client"]
    rtm_client = payload["rtm_client"]
    if "text" in data and "Hello" in data.get("text", []):
        channel_id = data["channel"]
        thread_ts = data["ts"]
        user = data["user"]

        try:
            response = await web_client.chat_postMessage(
                channel=channel_id, text=f"Hi <@{user}>!", thread_ts=thread_ts
            )
            print("ok", response)
        except SlackApiError as e:
            # You will get a SlackApiError if "ok" is False
            assert e.response["ok"] is False
            assert e.response["error"]  # str like 'invalid_auth', 'channel_not_found'
            print(f"Got an error: {e.response['error']}")


if __name__ == "__main__":
    import asyncio

    rtm_client = RTMClient(token=os.environ["SLACK_API_TOKEN"], run_async=True)
    loop = asyncio.get_event_loop()
    loop.set_debug(True)
    loop.run_until_complete(rtm_client.start())

x2check.py

import asyncio


async def hello():
    await asyncio.sleep(0.1)
    print("hello")


# これは OK
loop = asyncio.get_event_loop()
loop.run_until_complete(hello())
# これも OK
# asyncio.run(hello())

# これは OK
loop = asyncio.get_event_loop()
loop.run_until_complete(asyncio.gather(hello(), hello()))
# # これは NG
# asyncio.run(asyncio.gather(hello(), hello()))
# # ValueError: a coroutine was expected, got <_GatheringFuture pending>

x3hello.py

import os
from slack import RTMClient
from slack.errors import SlackApiError


@RTMClient.run_on(event="message")
async def say_hello(**payload):
    data = payload["data"]
    web_client = payload["web_client"]
    rtm_client = payload["rtm_client"]
    if "text" in data and "Hello" in data.get("text", []):
        channel_id = data["channel"]
        thread_ts = data["ts"]
        user = data["user"]

        try:
            response = await web_client.chat_postMessage(
                channel=channel_id, text=f"Hi <@{user}>!", thread_ts=thread_ts
            )
            print("ok", response)
        except SlackApiError as e:
            # You will get a SlackApiError if "ok" is False
            assert e.response["ok"] is False
            assert e.response["error"]  # str like 'invalid_auth', 'channel_not_found'
            print(f"Got an error: {e.response['error']}")


if __name__ == "__main__":
    import asyncio

    rtm_client = RTMClient(token=os.environ["SLACK_API_TOKEN"], run_async=True)

    async def run():
        await rtm_client.start()

    asyncio.run(run(), debug=True)

x4check.py

import asyncio


class Client:
    def __init__(self, loop=None):
        self.loop = loop or asyncio.get_event_loop()

    async def start(self):
        await asyncio.sleep(0.1, loop=self.loop)  # テキトーな処理
        print("started!")


asyncio.run(Client().start())

x5hello.py

import os
from slack import RTMClient
from slack.errors import SlackApiError


@RTMClient.run_on(event="message")
async def say_hello(**payload):
    data = payload["data"]
    web_client = payload["web_client"]
    rtm_client = payload["rtm_client"]
    if "text" in data and "Hello" in data.get("text", []):
        channel_id = data["channel"]
        thread_ts = data["ts"]
        user = data["user"]

        try:
            response = await web_client.chat_postMessage(
                channel=channel_id, text=f"Hi <@{user}>!", thread_ts=thread_ts
            )
            print("ok", response)
        except SlackApiError as e:
            # You will get a SlackApiError if "ok" is False
            assert e.response["ok"] is False
            assert e.response["error"]  # str like 'invalid_auth', 'channel_not_found'
            print(f"Got an error: {e.response['error']}")


if __name__ == "__main__":
    import asyncio

    async def run():
        loop = asyncio.get_event_loop()
        rtm_client = RTMClient(
            token=os.environ["SLACK_API_TOKEN"], loop=loop, run_async=True
        )
        await rtm_client.start()

    asyncio.run(run(), debug=True)