Skip to content


Subversion checkout URL

You can clone with
Download ZIP
Branch: master
Fetching contributors…

Cannot retrieve contributors at this time

200 lines (173 sloc) 6.67 KB
#!/usr/bin/env python
# -*- coding: utf-8 -*-
__version__ = '1.04'
__author__ = 'Liao Xuefeng ('
Python client SDK for sina weibo API using OAuth 2.
import json
except ImportError:
import simplejson as json
import time
import urllib
import urllib2
import logging
def _obj_hook(pairs):
convert json object to python object.
o = JsonObject()
for k, v in pairs.iteritems():
o[str(k)] = v
return o
class APIError(StandardError):
raise APIError if got failed json message.
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)
class JsonObject(dict):
general json object that can bind any fields but also act as a dict.
def __getattr__(self, attr):
return self[attr]
def __setattr__(self, attr, value):
self[attr] = value
def _encode_params(**kw):
Encode parameters.
args = []
for k, v in kw.iteritems():
qv = v.encode('utf-8') if isinstance(v, unicode) else str(v)
args.append('%s=%s' % (k, urllib.quote(qv)))
return '&'.join(args)
def _encode_multipart(**kw):
Build a multipart/form-data body with generated random boundary.
boundary = '----------%s' % hex(int(time.time() * 1000))
data = []
for k, v in kw.iteritems():
data.append('--%s' % boundary)
if hasattr(v, 'read'):
# file-like object:
ext = ''
filename = getattr(v, 'name', '')
n = filename.rfind('.')
if n != (-1):
ext = filename[n:].lower()
content =
data.append('Content-Disposition: form-data; name="%s"; filename="hidden"' % k)
data.append('Content-Length: %d' % len(content))
data.append('Content-Type: %s\r\n' % _guess_content_type(ext))
data.append('Content-Disposition: form-data; name="%s"\r\n' % k)
data.append(v.encode('utf-8') if isinstance(v, unicode) else v)
data.append('--%s--\r\n' % boundary)
return '\r\n'.join(data), boundary
_CONTENT_TYPES = { '.png': 'image/png', '.gif': 'image/gif', '.jpg': 'image/jpeg', '.jpeg': 'image/jpeg', '.jpe': 'image/jpeg' }
def _guess_content_type(ext):
return _CONTENT_TYPES.get(ext, 'application/octet-stream')
def _http_get(url, authorization=None, **kw):'GET %s' % url)
return _http_call(url, _HTTP_GET, authorization, **kw)
def _http_post(url, authorization=None, **kw):'POST %s' % url)
return _http_call(url, _HTTP_POST, authorization, **kw)
def _http_upload(url, authorization=None, **kw):'MULTIPART POST %s' % url)
return _http_call(url, _HTTP_UPLOAD, authorization, **kw)
def _http_call(url, method, authorization, **kw):
send an http request and expect to return a json object if no error.
params = None
boundary = None
if method==_HTTP_UPLOAD:
params, boundary = _encode_multipart(**kw)
params = _encode_params(**kw)
http_url = '%s?%s' % (url, params) if method==_HTTP_GET else url
http_body = None if method==_HTTP_GET else params
req = urllib2.Request(http_url, data=http_body)
if authorization:
req.add_header('Authorization', 'OAuth2 %s' % authorization)
if boundary:
req.add_header('Content-Type', 'multipart/form-data; boundary=%s' % boundary)
resp = urllib2.urlopen(req)
body =
r = json.loads(body, object_hook=_obj_hook)
if hasattr(r, 'error_code'):
raise APIError(r.error_code, getattr(r, 'error', ''), getattr(r, 'request', ''))
return r
class HttpObject(object):
def __init__(self, client, method):
self.client = client
self.method = method
def __getattr__(self, attr):
def wrap(**kw):
if self.client.is_expires():
raise APIError('21327', 'expired_token', attr)
return _http_call('%s%s.json' % (self.client.api_url, attr.replace('__', '/')), self.method, self.client.access_token, **kw)
return wrap
class APIClient(object):
API client using synchronized invocation.
def __init__(self, app_key, app_secret, redirect_uri=None, response_type='code', domain='', version='2'):
self.client_id = app_key
self.client_secret = app_secret
self.redirect_uri = redirect_uri
self.response_type = response_type
self.auth_url = 'https://%s/oauth2/' % domain
self.api_url = 'https://%s/%s/' % (domain, version)
self.access_token = None
self.expires = 0.0
self.get = HttpObject(self, _HTTP_GET) = HttpObject(self, _HTTP_POST)
self.upload = HttpObject(self, _HTTP_UPLOAD)
def set_access_token(self, access_token, expires_in):
self.access_token = str(access_token)
self.expires = float(expires_in)
def get_authorize_url(self, redirect_uri=None, display='default'):
return the authroize url that should be redirect.
redirect = redirect_uri if redirect_uri else self.redirect_uri
if not redirect:
raise APIError('21305', 'Parameter absent: redirect_uri', 'OAuth2 request')
return '%s%s?%s' % (self.auth_url, 'authorize', \
_encode_params(client_id = self.client_id, \
response_type = 'code', \
display = display, \
redirect_uri = redirect))
def request_access_token(self, code, redirect_uri=None):
return access token as object: {"access_token":"your-access-token","expires_in":12345678}, expires_in is standard unix-epoch-time
redirect = redirect_uri if redirect_uri else self.redirect_uri
if not redirect:
raise APIError('21305', 'Parameter absent: redirect_uri', 'OAuth2 request')
r = _http_post('%s%s' % (self.auth_url, 'access_token'), \
client_id = self.client_id, \
client_secret = self.client_secret, \
redirect_uri = redirect, \
code = code, grant_type = 'authorization_code')
r.expires_in += int(time.time())
return r
def is_expires(self):
return not self.access_token or time.time() > self.expires
def __getattr__(self, attr):
return getattr(self.get, attr)
Jump to Line
Something went wrong with that request. Please try again.