该代码参考新浪python api,适用于各个开源api请求

复制代码 代码如下:
# -*- coding: utf-8 -*-
import collections
import gzip
import urllib
import urllib2
from urlparse import urlparse

try:
    from cStringIO import StringIO
except ImportError:
    from StringIO import StringIO

try:
    import json
except ImportError:
    import simplejson as json

__author__ = \’myth\’

_HTTP_GET = 1
_HTTP_POST = 2
# 超时时间（秒）
TIMEOUT = 45
RETURN_TYPE = {\”json\”: 0, \”xml\”: 1, \”html\”: 2, \”text\”: 3}
_METHOD_MAP = {\’GET\’: _HTTP_GET, \’POST\’: _HTTP_POST}

class APIError(StandardError):
    \”\”\”
    raise APIError if receiving json message indicating failure.
    \”\”\”
    def __init__(self, error_code, error, request):
        self.error_code = error_code
        self.error = error
        self.request = request
        StandardError.__init__(self, error)

    def __str__(self):
        return \’APIError: %s: %s, request: %s\’ % (self.error_code, self.error, self.request)

# def callback_type(return_type=\’json\’):
#
#     default_type = \”json\”
#     default_value = RETURN_TYPE.get(default_type)
#     if return_type:
#         if isinstance(return_type, (str, unicode)):
#             default_value = RETURN_TYPE.get(return_type.lower(), default_value)
#     return default_value

def _format_params(_pk=None, encode=None, **kw):
    \”\”\”

    :param kw:
    :type kw:
    :param encode:
    :type encode:
    :return:
    :rtype:
    \”\”\”

    __pk = \’%s[%%s]\’ % _pk if _pk else \’%s\’
    encode = encode if encode else urllib.quote

    for k, v in kw.iteritems():
        _k = __pk % k
        if isinstance(v, basestring):
            qv = v.encode(\’utf-8\’) if isinstance(v, unicode) else v
            _v = encode(qv)
            yield _k, _v
        elif isinstance(v, collections.Iterable):
            if isinstance(v, dict):
                for _ck, _cv in _format_params(_pk=_k, encode=encode, **v):
                    yield _ck, _cv
            else:
                for i in v:
                    qv = i.encode(\’utf-8\’) if isinstance(i, unicode) else str(i)
                    _v = encode(qv)
                    yield _k, _v
        else:
            qv = str(v)
            _v = encode(qv)
            yield _k, _v

def encode_params(**kw):
    \”\”\”
    do url-encode parameters

    >>> encode_params(a=1, b=\’R&D\’)
    \’a=1&b=R%26D\’
    >>> encode_params(a=u\’\\u4e2d\\u6587\’, b=[\’A\’, \’B\’, 123])
    \’a=%E4%B8%AD%E6%96%87&b=A&b=B&b=123\’

    >>> encode_params(**{
        \’a1\’: {\’aa1\’: 1, \’aa2\’: {\’aaa1\’: 11}},
        \’b1\’: [1, 2, 3, 4],
        \’c1\’: {\’cc1\’: \’C\’, \’cc2\’: [\’Q\’, 1, \’@\’], \’cc3\’: {\’ccc1\’: [\’s\’, 2]}}
    })
    \’a1[aa1]=1&a1[aa2][aaa1]=11&c1[cc1]=C&c1[cc3][ccc1]=s&c1[cc3][ccc1]=2&c1[cc2]=Q&c1[cc2]=1&c1[cc2]=%40&b1=1&b1=2&b1=3&b1=4\’
    \”\”\”
    # args = []
    # for k, v in kw.iteritems():
    #     if isinstance(v, basestring):
    #         qv = v.encode(\’utf-8\’) if isinstance(v, unicode) else v
    #         args.append(\’%s=%s\’ % (k, urllib.quote(qv)))
    #     elif isinstance(v, collections.Iterable):
    #         for i in v:
    #             qv = i.encode(\’utf-8\’) if isinstance(i, unicode) else str(i)
    #             args.append(\’%s=%s\’ % (k, urllib.quote(qv)))
    #     else:
    #         qv = str(v)
    #         args.append(\’%s=%s\’ % (k, urllib.quote(qv)))
    # return \’&\’.join(args)

    args = []
    _urlencode = kw.pop(\’_urlencode\’, urllib.quote)
    for k, v in _format_params(_pk=None, encode=_urlencode, **kw):
        args.append(\’%s=%s\’ % (k, v))

    args = sorted(args, key=lambda s: s.split(\”=\”)[0])
    return \’&\’.join(args)

def _read_body(obj):
    using_gzip = obj.headers.get(\’Content-Encoding\’, \’\’) == \’gzip\’
    body = obj.read()
    if using_gzip:
        gzipper = gzip.GzipFile(fileobj=StringIO(body))
        fcontent = gzipper.read()
        gzipper.close()
        return fcontent
    return body

class JsonDict(dict):
    \”\”\”
    general json object that allows attributes to be bound to and also behaves like a dict
    \”\”\”

    def __getattr__(self, attr):
        try:
            return self[attr]
        except KeyError:
            raise AttributeError(r\”\’JsonDict\’ object has no attribute \’%s\’\” % attr)

    def __setattr__(self, attr, value):
        self[attr] = value

def _parse_json(s):
    \”\”\”
    parse str into JsonDict
    \”\”\”

    def _obj_hook(pairs):
        \”\”\”
        convert json object to python object
        \”\”\”
        o = JsonDict()
        for k, v in pairs.iteritems():
            o[str(k)] = v
        return o
    return json.loads(s, object_hook=_obj_hook)

def _parse_xml(s):
    \”\”\”
    parse str into xml
    \”\”\”

    raise NotImplementedError()

def _parse_html(s):
    \”\”\”
    parse str into html
    \”\”\”

    raise NotImplementedError()

def _parse_text(s):
    \”\”\”
    parse str into text
    \”\”\”

    raise NotImplementedError()

def _http_call(the_url, method, return_type=\”json\”, request_type=None, request_suffix=None, headers={}, _timeout=30, **kwargs):
    \”\”\”
    the_url: 请求地址
    method 请求方法（get，post）
    return_type： 返回格式解析
    request_suffix： 请求地址的后缀，如jsp，net
    _timeout: 超时时间
    kwargs: 请求参数
    \”\”\”

    http_url = \”%s.%s\” (the_url, request_suffix) if request_suffix else the_url

    if request_type == \’json\’:
        headers[\’Content-Type\’] = \’application/json\’
        # method = _HTTP_POST
        # json_data = json.dumps(kwargs)
        # # convert str to bytes (ensure encoding is OK)
        # params = json_data.encode(\’utf-8\’)
        params = json.dumps(kwargs)
    else:
        params = encode_params(**kwargs)
        http_url = \’%s?%s\’ % (http_url, params) if method == _HTTP_GET else http_url
    print http_url
    # u = urlparse(http_url)
    # headers.setdefault(\”host\”, u.hostname)
    http_body = None if method == _HTTP_GET else params

    req = urllib2.Request(http_url, data=http_body, headers=headers)

    callback = globals().get(\’_parse_{0}\’.format(return_type))
    if not hasattr(callback, \’__call__\’):
        print \”return \’%s\’ unable to resolve\” % return_type
        callback = _parse_json
    try:

        resp = urllib2.urlopen(req, timeout=_timeout if _timeout else TIMEOUT)
        body = _read_body(resp)
        r = callback(body)
        # if hasattr(r, \’error_code\’):
        #     raise APIError(r.error_code, r.get(\’error\’, \’\’), r.get(\’request\’, \’\’))
        return r
    except urllib2.HTTPError, e:
        try:
            body = _read_body(e)
            r = callback(body)
            return r
        except:
            r = None
        # if hasattr(r, \’error_code\’):
        #     raise APIError(r.error_code, r.get(\’error\’, \’\’), r.get(\’request\’, \’\’))
            raise e

class HttpObject(object):

    def __init__(self, client, method):
        self.client = client
        self.method = method

    def __getattr__(self, attr):

        def wrap(**kw):
            if attr:
                the_url = \’%s/%s\’ % (self.client.api_url, attr.replace(\’__\’, \’/\’))
            else:
                the_url = self.client.api_url
            return _http_call(the_url, self.method, **kw)
        return wrap

    def __call__(self, **kw):
        return _http_call(self.client.api_url, self.method, **kw)

class APIClient(object):
    \”\”\”
    使用方法：
        比如：api 请求地址为：http://api.open.zbjdev.com/kuaiyinserv/kuaiyin/billaddress
             请求方式为： GET
             需要的参数为：user_id 用户的UID
                         is_all 是否查询所有数据，0为默认邮寄地址 1为全部邮寄地址
                         access_token 平台认证
             返回数据为：json

        那么此时使用如下：
        domain = \”api.open.zbjdev.com\”
        #如果是https请求，需要将is_https设置为True
        client = APIClient(domain)
        data = {\”user_id\”: \”14035462\”, \”is_all\”: 1, \”access_token\”: \”XXXXXXXXXX\”}
        # 如果是post请求，请将get方法改为post方法
        result = client.kuaiyinserv.kuaiyin.billaddress.get(return_type=\”json\”, **data)
        #等同于
        # result = client.kuaiyinserv__kuaiyin__billaddress__get(return_type=\”json\”, **data)
        # result = client.kuaiyinserv__kuaiyin__billaddress(return_type=\”json\”, **data)
    \”\”\”

    def __init__(self, domain, is_https=False):

        http = \”http\”
        if domain.startswith(\”http://\”) or domain.startswith(\”https://\”):
            http, domain = domain.split(\”://\”)
        # else:
        if is_https:
            http = \”https\”

        self.api_url = (\’%s://%s\’ % (http, domain)).rstrip(\”/\”)
        self.get = HttpObject(self, _HTTP_GET)
        self.post = HttpObject(self, _HTTP_POST)

    def __getattr__(self, attr):
        if \’__\’ in attr:

            method = self.get
            if attr[-6:] == \”__post\”:
                attr = attr[:-6]
                method = self.post

            elif attr[-5:] == \”__get\”:
                attr = attr[:-5]

            if attr[:2] == \’__\’:
                attr = attr[2:]
            return getattr(method, attr)
        return _Callable(self, attr)

class _Executable(object):

    def __init__(self, client, method, path):
        self._client = client
        self._method = method
        self._path = path

    def __call__(self, **kw):
        method = _METHOD_MAP[self._method]
        return _http_call(\’%s/%s\’ % (self._client.api_url, self._path), method, **kw)

    def __str__(self):
        return \’_Executable (%s %s)\’ % (self._method, self._path)

    __repr__ = __str__

class _Callable(object):

    def __init__(self, client, name):
        self._client = client
        self._name = name

    def __getattr__(self, attr):
        if attr == \’get\’:
            return _Executable(self._client, \’GET\’, self._name)
        if attr == \’post\’:
            return _Executable(self._client, \’POST\’, self._name)
        name = \’%s/%s\’ % (self._name, attr)
        return _Callable(self._client, name)

    def __str__(self):
        return \’_Callable (%s)\’ % self._name

    __repr__ = __str__

def test_logistics():

    domain = \”https://api.kuaidi100.com/api\”

    #如果是https请求，需要将is_https设置为True
    client = APIClient(domain)

    data = {\”id\”: \”45f2d1f2sds\”, \”com\”: \”yunda\”, \”nu\”: \”1500066330925\”}
    result = client.__get(_timeout=2, **data)
    print result
    print result[\”message\”]
    print result.get(\”message\”)
    print result.message

if __name__ == \’__main__\’:

    # test_logistics()

    data = {
        \”data\”:{
            \”id\”: 1,
            \”pid\”: 3,
            \”name\”: u\’中的阿斯顿\’
        },
        \”sign\”: \’asdfdsdsfsdf\’,
    }

    domain = \”kuaiyin.zhubajie.com\”

    client = APIClient(domain)
    # headers = {\’Content-Type\’: \’application/json\’}
    headers = {\”host\”: domain}

    result = client.api.zhubajie.fz.info.post(return_type=\”json\”, request_type=\”json\”, headers=headers, **data)
    print result
    print result[\’data\’][\’msg\’]

    c = APIClient(\’task.6.zbj.cn\’)
    r = getattr(c.api, \’kuaiyin-action-delcache\’).post(request_type=\”json\”,
                                                       headers={},
                                                       **{\”sign\”: \”\”,
                                                          \”data\”: {
                                                              \”product\”: \”pvc\”,
                                                              \”category\”: \”card\”,
                                                              \”req_type\”: \”pack_list\”
                                                          }})
    print r
    print r[\’data\’][\’msg\’]