Skip to content

Instantly share code, notes, and snippets.

@exhuma
Last active March 18, 2018 19:01
Show Gist options
  • Save exhuma/0c28a4e41c476e122909a3521ae96d1a to your computer and use it in GitHub Desktop.
Save exhuma/0c28a4e41c476e122909a3521ae96d1a to your computer and use it in GitHub Desktop.
Quick & Dirty Rocket.Chat client

Simple Rocket Chat Client

About

This is a really simple module offering a thin wrapper around the Rocket.Chat REST API. It only provides a simple means to deal with authentication and does not provide any functions for the REST endpoints.

Installation

# Clone & Create Virtual environment
$ git clone https://gist.github.com/exhuma/0c28a4e41c476e122909a3521ae96d1a rchat-client
$ cd rchat-client
$ python -m venv env

# Install the requirements into the new environment
$ ./env/bin/pip install -r requirements.txt

After running the above steps the Python interpreter in env/bin/python will be able to run the main script.

Usage

See the function main in main.py.

API calls are rooted at /api/v1.

For API documentation see https://rocket.chat/docs/developer-guides/rest-api/

Configuration

Place a file called app.py into ~/.config/ept/rchat with the following contents (using your own values of course):

[server]
url=https://url.to.your.rocket.chat

[auth]
login=your-username
password=your-password

Given that the file contains sensitive information, it will only be loaded if it has 600 permissions!

import logging
from contextlib import contextmanager
import requests
from config_resolver import SecuredConfig
LOG = logging.getLogger(__name__)
class Client:
def __init__(self, url, token, user_id):
self.token = token
self.user_id = user_id
self.url = url + '/api/v1'
self.__headers = {
'X-Auth-Token': token,
'X-User-Id': user_id,
}
def post(self, url, *args, **kwargs):
headers = {**kwargs.pop('headers', {}), **self.__headers}
kwargs['headers'] = headers
return requests.post(self.url + url, *args, **kwargs)
def get(self, url, *args, **kwargs):
headers = {**kwargs.pop('headers', {}), **self.__headers}
kwargs['headers'] = headers
return requests.get(self.url + url, *args, **kwargs)
@contextmanager
def session(url, username, password):
response = requests.post(url + '/api/v1/login',
{'username': username, 'password': password})
data = response.json()['data']
auth_client = Client(url, data['authToken'], data['userId'])
yield auth_client
auth_client.post('/api/v1/logout')
def main():
logging.basicConfig(level=0)
config = SecuredConfig('ept', 'rchat', require_load=True)
url = config.get('server', 'url')
login = config.get('auth', 'login')
password = config.get('auth', 'password')
with session(url, login, password) as client:
data = {
'channel': '@mike',
'text': '`hello world!`',
'emoji': ':robot:'
}
client.post('/chat.postMessage', data=data)
if __name__ == '__main__':
main()
certifi==2018.1.18
chardet==3.0.4
config-resolver==4.2.4
idna==2.6
pipfile==0.0.2
pkg-resources==0.0.0
requests==2.18.4
toml==0.9.4
urllib3==1.22
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment