leancloud.client 源代码

# coding: utf-8

from __future__ import absolute_import
from __future__ import division
from __future__ import print_function
from __future__ import unicode_literals

import os
import json
import time
import hashlib
import functools

import six
import requests

import leancloud
from leancloud import utils
from leancloud.app_router import AppRouter

__author__ = 'asaka <lan@leancloud.rocks>'


APP_ID = None
APP_KEY = None
MASTER_KEY = None
HOOK_KEY = None
USE_PRODUCTION = '1'
USE_HTTPS = True
# 兼容老版本,如果 USE_MASTER_KEY 为 None ,并且 MASTER_KEY 不为 None,则使用 MASTER_KEY
# 否则依据 USE_MASTER_KEY 来决定是否使用 MASTER_KEY
USE_MASTER_KEY = None
REGION = 'CN'

app_router = None
session = requests.Session()
request_hooks = {}

SERVER_VERSION = '1.1'

TIMEOUT_SECONDS = 15


[文档]def init(app_id, app_key=None, master_key=None, hook_key=None): """初始化 LeanCloud 的 AppId / AppKey / MasterKey :type app_id: string_types :param app_id: 应用的 Application ID :type app_key: None or string_types :param app_key: 应用的 Application Key :type master_key: None or string_types :param master_key: 应用的 Master Key :param hook_key: application's hook key :type hook_key: None or string_type """ if (not app_key) and (not master_key): raise RuntimeError('app_key or master_key must be specified') global APP_ID, APP_KEY, MASTER_KEY, HOOK_KEY APP_ID = app_id APP_KEY = app_key MASTER_KEY = master_key if hook_key: HOOK_KEY = hook_key else: HOOK_KEY = os.environ.get('LEANCLOUD_APP_HOOK_KEY')
def need_init(func): @functools.wraps(func) def new_func(*args, **kwargs): if APP_ID is None: raise RuntimeError('LeanCloud SDK must be initialized') headers = { 'Content-Type': 'application/json;charset=utf-8', 'X-LC-Id': APP_ID, 'X-LC-Prod': USE_PRODUCTION, 'User-Agent': 'AVOS Cloud python-{0} SDK'.format(leancloud.__version__), } md5sum = hashlib.md5() current_time = six.text_type(int(time.time() * 1000)) if (USE_MASTER_KEY is None and MASTER_KEY) or USE_MASTER_KEY is True: md5sum.update((current_time + MASTER_KEY).encode('utf-8')) headers['X-LC-Sign'] = md5sum.hexdigest() + ',' + current_time + ',master' else: # In python 2.x, you can feed this object with arbitrary # strings using the update() method, but in python 3.x, # you should feed with bytes-like objects. md5sum.update((current_time + APP_KEY).encode('utf-8')) headers['X-LC-Sign'] = md5sum.hexdigest() + ',' + current_time user = leancloud.User.get_current() if user: headers['X-LC-Session'] = user._session_token return func(headers=headers, *args, **kwargs) return new_func def get_url(part): # try to use the base URL from environ url = os.environ.get('LC_API_SERVER') or os.environ.get('LEANCLOUD_API_SERVER') if url: return '{}/{}{}'.format(url, SERVER_VERSION, part) global app_router if app_router is None: app_router = AppRouter(APP_ID, REGION) if part.startswith('/push') or part.startswith('/installations'): host = app_router.get('push') elif part.startswith('/collect'): host = app_router.get('stats') elif part.startswith('/functions') or part.startswith('/call'): host = app_router.get('engine') else: host = app_router.get('api') r = { 'schema': 'https' if USE_HTTPS else 'http', 'version': SERVER_VERSION, 'host': host, 'part': part, } return '{schema}://{host}/{version}{part}'.format(**r)
[文档]def use_production(flag): """调用生产环境 / 开发环境的 cloud func / cloud hook 默认调用生产环境。 """ global USE_PRODUCTION USE_PRODUCTION = '1' if flag else '0'
[文档]def use_master_key(flag=True): """是否使用 master key 发送请求。 如果不调用此函数,会根据 leancloud.init 的参数来决定是否使用 master key。 :type flag: bool """ global USE_MASTER_KEY if not flag: USE_MASTER_KEY = False return if not MASTER_KEY: raise RuntimeError('LeanCloud SDK master key not specified') USE_MASTER_KEY = True
def check_error(func): @functools.wraps(func) def new_func(*args, **kwargs): response = func(*args, **kwargs) assert isinstance(response, requests.Response) if response.headers.get('Content-Type') == 'text/html': raise leancloud.LeanCloudError(-1, 'Bad Request') content = response.json() if 'error' in content: raise leancloud.LeanCloudError(content.get('code', 1), content.get('error', 'Unknown Error')) return response return new_func
[文档]def use_region(region): if region not in ('CN', 'US'): raise ValueError('currently no nodes in the region') global REGION REGION = region
def get_server_time(): response = check_error(session.get)(get_url('/date'), timeout=TIMEOUT_SECONDS) return utils.decode('iso', response.json()) def get_app_info(): return { 'app_id': APP_ID, 'app_key': APP_KEY, 'master_key': MASTER_KEY, 'hook_key': HOOK_KEY, } @need_init @check_error def get(url, params=None, headers=None): if not params: params = {} else: for k, v in six.iteritems(params): if isinstance(v, dict): params[k] = json.dumps(v, separators=(',', ':')) response = session.get( get_url(url), headers=headers, params=params, timeout=TIMEOUT_SECONDS, hooks=request_hooks, ) return response @need_init @check_error def post(url, params, headers=None): response = session.post( get_url(url), headers=headers, data=json.dumps(params, separators=(',', ':')), timeout=TIMEOUT_SECONDS, hooks=request_hooks, ) return response @need_init @check_error def put(url, params, headers=None): response = session.put( get_url(url), headers=headers, data=json.dumps(params, separators=(',', ':')), timeout=TIMEOUT_SECONDS, hooks=request_hooks, ) return response @need_init @check_error def delete(url, params=None, headers=None): response = session.delete( get_url(url), headers=headers, data=json.dumps(params, separators=(',', ':')), timeout=TIMEOUT_SECONDS, hooks=request_hooks, ) return response