slack.io.aiohttp - Aiohttp implementation

Examples

Query

import os
import sys
import pprint
import asyncio
import logging

import aiohttp
import slack
from slack.io.aiohttp import SlackAPI

logging.basicConfig(level=logging.INFO)
LOG = logging.getLogger(__name__)


async def query(client):

    data = await client.query(slack.methods.AUTH_TEST)
    pprint.pprint(data)


if __name__ == "__main__":

    if len(sys.argv) > 1:
        TOKEN = sys.argv[1]
    else:
        TOKEN = os.environ.get("SLACK_TOKEN")

    if not TOKEN:
        raise ValueError("No slack token provided !")

    loop = asyncio.get_event_loop()
    session = aiohttp.ClientSession(loop=loop)
    slack_client = SlackAPI(token=TOKEN, session=session)

    try:
        loop.run_until_complete(query(slack_client))
    except KeyboardInterrupt:
        pass

    loop.run_until_complete(session.close())

Iterate

import os
import sys
import pprint
import asyncio
import logging

import aiohttp
import slack
from slack.io.aiohttp import SlackAPI

logging.basicConfig(level=logging.INFO)
LOG = logging.getLogger(__name__)


async def iterate(client):

    async for channel in client.iter(slack.methods.CHANNELS_LIST, limit=4):
        pprint.pprint(channel)


if __name__ == "__main__":

    if len(sys.argv) > 1:
        TOKEN = sys.argv[1]
    else:
        TOKEN = os.environ.get("SLACK_TOKEN")

    if not TOKEN:
        raise ValueError("No slack token provided !")

    loop = asyncio.get_event_loop()
    session = aiohttp.ClientSession(loop=loop)
    slack_client = SlackAPI(token=TOKEN, session=session)

    try:
        loop.run_until_complete(iterate(slack_client))
    except KeyboardInterrupt:
        pass

    loop.run_until_complete(session.close())

RTM

import os
import sys
import pprint
import asyncio
import logging

import aiohttp
import slack
from slack.events import Message
from slack.io.aiohttp import SlackAPI

logging.basicConfig(level=logging.INFO)
LOG = logging.getLogger(__name__)


async def rtm(client):

    async for event in client.rtm():
        pprint.pprint(event)

        if isinstance(event, Message):
            asyncio.ensure_future(respond_to_message(event, client))


async def respond_to_message(message, client):
    response = message.response()
    response["text"] = "Hello world !"
    await client.query(slack.methods.CHAT_POST_MESSAGE, data=response.serialize())


if __name__ == "__main__":

    if len(sys.argv) > 1:
        TOKEN = sys.argv[1]
    else:
        TOKEN = os.environ.get("SLACK_TOKEN")

    if not TOKEN:
        raise ValueError("No slack token provided !")

    loop = asyncio.get_event_loop()
    session = aiohttp.ClientSession(loop=loop)
    slack_client = SlackAPI(token=TOKEN, session=session)

    try:
        loop.run_until_complete(rtm(slack_client))
    except KeyboardInterrupt:
        pass

    loop.run_until_complete(session.close())