Skip to content

Instantly share code, notes, and snippets.

@podhmo

podhmo/x0hello.py

Created Jul 10, 2020
Embed
What would you like to do?
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)
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())
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>
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)
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())
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)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment