A nice way to generate external apis

README.md

I hate using third party language libraries because they are often just a crappier more OOP'd version of a nice rest api. Here's a sample framework for automatically generating call chains that map to the RESTful url.

Just define the verbs and nouns you want to use in the resources set.

"POST /buttons" maps to "api.buttons.post"

or

"GET accounts/:account_id" maps to "api.accounts.id(account_id).get"

Example usage

response = yield gen.Task(self.backend.coinbase.buttons.post, **{
      "account_id": options.coinbase_account_id,
      "button": {
        "name": "Fund Journalism",
        "type": "buy_now",
        "subscription": False,
        "price_string": "0.0105",
        "price_currency_iso": "BTC",
        "custom": bitcoin_topic.hash(),
        "description": "Testing",
        "callback_url": callback_url,
        "variable_price": True,
        "choose_price": True,
        "custom_secure": True
      }
    })

gistfile1.py

# -*- coding: utf-8 -*-

# Copyright 2014 Dmitri Cherniak
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

__all__ = ['Coinbase']

import urllib
import functools
import hashlib
import hmac
import time
import json

from tornado import httpclient, escape
from tornado.options import define, options

define("coinbase_api_secret")
define("coinbase_api_key")

class Coinbase(object):
    api_hostname = 'coinbase.com/api'
    api_version = 'v1'

    resources = set([
        'buttons',
        'account_changes',
        'account',
        'balance',
        'primary',
        'addresses',
        'create_order',
        'orders',
        'buys',
        'contacts',
        'currencies',
        'exchange_rates',
        'orders',
        'recurring_payments',
        'reports',
        'sells',
        'users'
    ])

    def __init__(self, api_key=None, api_secret=None, blocking=False):
        self.api_key  = api_key if api_key else options.coinbase_api_key
        self.api_secret  = api_secret if api_secret else options.coinbase_api_secret
        self.blocking = blocking
        self.url = None

        if blocking:
            self.httpclient_instance = httpclient.HTTPClient()
        else:
            self.httpclient_instance = httpclient.AsyncHTTPClient()


    def __getattr__(self, name):
        '''
        Builds API URL.
        Example:
            tornado_coinbase.Coinbase().buttons.post
        '''
        if name in self.__class__.resources:
            self.url = '/'.join([self.url or self.api_endpoint, name])
            return self
        else:
            raise AttributeError(name)


    def __repr__(self):
        return "%s(api_key=%s, url=%s)" % (self.__class__.__name__, self.api_key, self.url)


    @property
    def api_endpoint(self):
        return 'https://%s/%s' % (self.__class__.api_hostname, self.__class__.api_version)


    def id(self, id):
        '''
        Append ID to constructed URL.
        Example:
            tornado_coinbase.Coinbase().buttons.id('code').orders.get(callback=lambda x: x)
        '''
        self.url = '/'.join([self.url or self.api_endpoint, str(id)])
        return self


    def reset_url(self):
        self.url = None


    def get(self, **kwargs):
        return self._call_check_blocking_first('GET', **kwargs)


    def post(self, **kwargs):
        return self._call_check_blocking_first('POST', **kwargs)


    def put(self, **kwargs):
        return self._call_check_blocking_first('PUT', **kwargs)


    def delete(self, **kwargs):
        return self._call_check_blocking_first('DELETE', **kwargs)


    def _call_check_blocking_first(self, http_method, **kwargs):
        if self.blocking:
            http_response = self._call(http_method, **kwargs)
            return self._parse_response(None, http_response)
        else:
            return self._call(http_method, **kwargs)

    def _generate_headers(self, url, body=None):
        nonce = int(time.time() * 1e6)
        message = str(nonce) + url + ('' if body is None else body)
        signature = hmac.new(self.api_secret.encode(), message.encode(), hashlib.sha256).hexdigest()

        headers = {
            'ACCESS_KEY': self.api_key,
            'ACCESS_SIGNATURE': signature,
            'ACCESS_NONCE': str(nonce),
            'Accept': 'application/json'
        }
        
        if body:
            headers.update({'Content-Type':'application/json'})
        
        return headers
    
    def _call(self, http_method, callback=None, **kwargs):
        copy_of_url = self.url

        # reset self.url
        self.reset_url()

        httpclient_args   = [copy_of_url]
        
        httpclient_kwargs = {
            'method': http_method
        }

        if http_method == 'GET' or http_method == 'DELETE' and kwargs:
            httpclient_kwargs['headers'] = self._generate_headers(copy_of_url)
            httpclient_args = [copy_of_url + '?' + urllib.urlencode(self._nested_dict_to_url(kwargs))]

        elif http_method != 'GET' and kwargs:
            body = json.dumps(kwargs)
            httpclient_kwargs['headers'] = self._generate_headers(copy_of_url, body=body)
            httpclient_kwargs['body'] = body

        if not self.blocking:
            httpclient_args.append(functools.partial(self._parse_response, callback))

        return self.httpclient_instance.fetch(*httpclient_args, **httpclient_kwargs)


    def _nested_dict_to_url(self, d):
        """
        We want post vars of form:
        {'foo': 'bar', 'nested': {'a': 'b', 'c': 'd'}}
        to become (pre url-encoding):
        foo=bar&nested[a]=b&nested[c]=d
        """
        stk = []
        for key, value in d.items():
            if isinstance(value, dict):
                n = {}
                for k, v in value.items():
                    n["%s[%s]" % (key, k)] = v
                stk.extend(self._nested_dict_to_url(n))
            else:
                stk.append((key, value))
        return stk


    def _parse_response(self, callback, response):
        """Parse a response from the API"""
        try:
            res = escape.json_decode(response.body)
        except Exception, e:
            e.args += ('API response was: %s' % response,)
            raise e

        if callback:
            callback(res)
        else:
            return res