etherpad_client.py

#!/usr/bin/env python3
"""Etherpad Lite client. Modified from https://github.com/Changaco/python-etherpad_lite"""

import requests
import functools
import json

if __name__ == '__main__':
    import argparse
    import getpass
    import shlex
    import os
    try:
        import readline
    except ImportError:
        pass


class EtherpadErrorCode:
    RESPONSE_FORMAT = -2
    REQUEST = -1
    OK = 0
    WRONG_PARAMETERS = 1
    INTERNAL = 2
    NO_SUCH_FUNCTION = 3
    API_KEY = 4


class EtherpadException(Exception):
    """generic error."""
    def __init__(self, message, code=None):
        super().__init__(message)
        self.code = code


class EtherpadClient:
    expected_response_keys = {'code', 'message', 'data'}
    def __init__(self, api_version: str, api_url: str, api_key: str):
        api_url += '/' * (not api_url.endswith('/'))
        self.url = '/'.join((api_url, 'api', api_version, ''))
        self.key = api_key

    def __call__(self, command, params):
        params = {'apikey': self.key, **params}
        try:
            req = requests.get(self.url + command, params)
        except requests.RequestException as e:
            raise EtherpadException(str(e), EtherpadErrorCode.REQUEST)
        try:
            resp = req.json()
        except json.JSONDecodeError as e:
            raise EtherpadException(str(e), EtherpadErrorCode.RESPONSE_FORMAT)
        if set(resp.keys()) != self.expected_response_keys:
            raise EtherpadException(
                'The response contained {}, while {} was expected'
                .format(set(resp.keys()), self.expected_response_keys),
                EtherpadErrorCode.RESPONSE_FORMAT)
        if resp['code']:
            raise EtherpadException(resp['message'], resp['code'])
        return resp['data']

    def __getattr__(self, name):
        return functools.partial(self, name)


# the following part is for interactive usage


class UserExit(Exception):
    pass


class UserError(Exception):
    pass


def get_api_key(value):
    try:
        with open(value) as f:
            return f.read().strip()
    except FileNotFoundError:
        return value


def get_args():
    parser = argparse.ArgumentParser()
    parser.add_argument('api_version', help='The API version to use')
    parser.add_argument('api_url', help='The API location, including scheme but ommitting "/api", e.g. http://localhost:9001')
    parser.add_argument('--prompt', default='Etherpad Lite APIv{api_version}::',
                        help='The prompt. May include {api_version}, {api_url}')
    parser.add_argument('--api-key', help='The API key. May be the key itself or a file to read the key from',
                        type=get_api_key)
    return parser.parse_args()


def read_input(prompt):
    try:
        ui = input(prompt)
    except EOFError:
        raise UserExit
    try:
        ui = shlex.split(ui)
    except ValueError as e:
        raise UserError(str(e))
    try:
        command, *args_text = ui
    except ValueError:
        raise UserError('command missing')
    arg_dict = {}
    for arg in args_text:
        try:
            name, value = arg.split('=', 1)
        except ValueError:
            raise UserError('"{}" is missing an "="')
        arg_dict[name] = value
    return command, arg_dict


def loop(client, prompt):
    while True:
        try:
            args = read_input(prompt)
        except UserError as e:
            print(e)
        except UserExit:
            print()
            return
        else:
            try:
                r = client(*args)
            except EtherpadException as e:
                print(e)
            else:
                print(r)


def main():
    args = get_args()
    if args.api_key is None:
        args.api_key = getpass.getpass('API key: ')
    client = EtherpadClient(args.api_version, args.api_url, args.api_key)
    prompt = args.prompt.format(**args.__dict__)
    loop(client, prompt)


if __name__ == '__main__':
    main()