# -*- coding: utf-8 -*-
from __future__ import absolute_import, unicode_literals
import sys
import inspect
try:
import simplejson as json
except ImportError:
import json
import requests
from teambition import api
from teambition.api.base import TeambitionAPI
from teambition.utils import JSONEncoder, JSONDecoder
from teambition.exceptions import TeambitionException
def _is_api_endpoint(obj):
return isinstance(obj, TeambitionAPI)
[文档]class Teambition(object):
"""
Teambition API 客户端
"""
API_BASE_URL = 'https://api.teambition.com/'
# API endpoints
oauth = api.OAuth()
""":doc:`oauth`"""
projects = api.Projects()
""":doc:`projects`"""
tasklists = api.Tasklists()
""":doc:`tasklists`"""
stages = api.Stages()
""":doc:`stages`"""
tasks = api.Tasks()
""":doc:`tasks`"""
users = api.Users()
""":doc:`users`"""
organizations = api.Organizations()
""":doc:`organizations`"""
stagetemplates = api.StageTemplates()
""":doc:`stagetemplates`"""
teams = api.Teams()
""":doc:`teams`"""
subtasks = api.Subtasks()
""":doc:`subtasks`"""
messages = api.Messages()
""":doc:`messages`"""
posts = api.Posts()
""":doc:`posts`"""
collections = api.Collections()
""":doc:`collections`"""
works = api.Works()
""":doc:`works`"""
events = api.Events()
""":doc:`events`"""
tags = api.Tags()
""":doc:`tags`"""
objectlinks = api.ObjectLinks()
""":doc:`objectlinks`"""
activities = api.Activities()
""":doc:`activities`"""
webhooks = api.Webhooks()
""":doc:`webhooks`"""
bookkeepings = api.BookKeepings()
""":doc:`bookkeepings`"""
entrycategories = api.EntryCategories()
"""doc:`entrycategories`"""
entries = api.Entries()
""":doc:`entries`"""
def __new__(cls, *args, **kwargs):
self = super(Teambition, cls).__new__(cls)
if sys.version_info[:2] == (2, 6):
# Python 2.6 inspect.gemembers bug workaround
# http://bugs.python.org/issue1785
for _class in cls.__mro__:
if issubclass(_class, Teambition):
for name, tb_api in _class.__dict__.items():
if isinstance(tb_api, TeambitionAPI):
api_cls = type(tb_api)
tb_api = api_cls(self)
setattr(self, name, tb_api)
else:
api_endpoints = inspect.getmembers(self, _is_api_endpoint)
for name, tb_api in api_endpoints:
api_cls = type(tb_api)
tb_api = api_cls(self)
setattr(self, name, tb_api)
return self
def __init__(self, client_id, client_secret, access_token=None):
"""
初始化 Teambition API Client
:param client_id: 申请应用时分配的 client_id
:param client_secret: 申请应用时分配的 client_secret
:param access_token: 可选,access_token
"""
self.client_id = client_id
self.client_secret = client_secret
self._access_token = access_token
def _request(self, method, endpoint, **kwargs):
if not endpoint.startswith(('http://', 'https://')):
api_base_url = kwargs.pop('api_base_url', self.API_BASE_URL)
url = '{base}{endpoint}'.format(
base=api_base_url,
endpoint=endpoint
)
else:
url = endpoint
if isinstance(kwargs.get('data', ''), dict):
body = json.dumps(
kwargs['data'],
ensure_ascii=False,
cls=JSONEncoder
)
body = body.encode('utf-8')
kwargs['data'] = body
if 'headers' not in kwargs:
kwargs['headers'] = {}
if 'Content-Type' not in kwargs['headers']:
kwargs['headers']['Content-Type'] = 'application/json'
if 'Authorization' not in kwargs['headers']:
kwargs['headers']['Authorization'] = 'OAuth2 {0}'.format(
self.access_token
)
res = requests.request(
method=method,
url=url,
**kwargs
)
# Error handling, reraise RequestException as TeambitionException
try:
res.raise_for_status()
except requests.RequestException as reqe:
code = -1
message = None
if reqe.response.text:
try:
errinfo = json.loads(reqe.response.text)
code = errinfo['code']
message = errinfo['message']
except ValueError:
pass
raise TeambitionException(
code=code,
message=message,
client=self,
request=reqe.request,
response=reqe.response
)
result = res.json(cls=JSONDecoder) if res.text else {}
return result
def get(self, endpoint, **kwargs):
return self._request('get', endpoint, **kwargs)
def post(self, endpoint, **kwargs):
return self._request('post', endpoint, **kwargs)
def put(self, endpoint, **kwargs):
return self._request('put', endpoint, **kwargs)
def delete(self, endpoint, **kwargs):
return self._request('delete', endpoint, **kwargs)
@property
def access_token(self):
"""
获取 access_token
:return: access_token
"""
return self._access_token
@access_token.setter
def access_token(self, token):
"""
设置 access_token
:param token: access_token
"""
self._access_token = token