python的简单rest-api客户端。
popget的Python项目详细描述
一个简单的无铃和哨声的rest-api客户机,可选地支持异步请求。
我们将此用于服务到服务异类中的请求 微服务环境。
用法
pip install popget
配置
设置主要通过python文件进行配置,例如 作为您现有的Django ^ {TT1}$。 要引导它,有几个env变量来控制如何配置 已加载:
- POPGET_APP_CONFIG 应该是python模块的导入路径,例如: POPGET_APP_CONFIG=django.conf.settings
- POPGET_CONFIG_NAMESPACE 设置用于从env和 配置文件。默认为POPGET。
有关详细信息,请参见popget/conf/defaults.py的源代码。
一些有用的配置键(它们都以前缀 POPGET_默认值:
- <namespace>_CLIENT_DEFAULT_USER_AGENT发出请求时,popget将使用此 字符串作为用户代理。
- <namespace>_CLIENT_TIMEOUT如果None则没有超时,否则此超时 (秒)将应用于所有请求。请求超时将 返回504响应,该响应将作为HTTPError引发。
apiclient
您将使用子类APIClient来生成您的api。你不需要 实例化客户端,所有方法都是类方法。
例如
frompopgetimportAPIClient,Arg,GetEndpointclassThingServiceClient(APIClient):classConfig:base_url='http://things.depop.com'get_things=GetEndpoint('/things/{user_id}/',# url format stringquerystring_args=(Arg('type',required=True),# required querystring param (validated on call)))
生成一个客户端方法,您可以调用如下:
data=ThingServiceClient.get_things(user_id=2345,type='cat')
它将执行如下请求:
GET http://things.depop.com/things/2345/?type=cat
如果响应是"Content-Type: application/json",则data是 已经反序列化。
在python 3中,str和bytes之间还有进一步的区别。 如果内容类型头包含text/,则返回值 将被编码为str(通过底层python-requestslib)。 其他内容类型将返回bytes。
我们使用raise_for_status,因此任何>;=400都将引发requests.HTTPError。
APIEndpoint
APIEndpoint是端点方法的基类。GetEndpoint, PostEndpoint,PutEndpoint,PatchEndpoint和DeleteEndpoint 为方便起见,允许省略方法arg。
来自url路径(格式字符串)、querystring和请求头的参数 (值部分的格式字符串)将被提取并可用 在客户端类上生成的可调用方法上显示为kwargs。
这意味着arg名称在所有三个arg源中必须是唯一的。 这是可行的,因为当 实现客户端(它们只是格式化字符串标识符,而不是 而不是rest api本身的一部分,比如querystring参数)。
例如
frompopgetimportAPIClient,Arg,GetEndpointclassThingServiceClient(APIClient):classConfig:base_url='http://things.depop.com'get_things=GetEndpoint('/things/{user_id}/',# url (format string)querystring_args=(Arg('type',required=True),Arg('offset_id'),Arg('limit',default=25),),request_headers={# added to all requests'Authorization':'Bearer {access_token}'# (format string)})
这将为您提供一个具有get_things方法的客户机,您可以像这样调用:
response_data=ThingServiceClient.get_things(user_id=123,type='cat',offset_id='65345ff34e344ab53c',limit=20,access_token='87a64c98b62d39e8625f',)
querystring参数可以有一个callable作为默认值,例如:
fromdatetimeimportdatetimefrompopgetimportAPIClient,Arg,GetEndpointdefnow():returndatetime.now().isoformat()classThingServiceClient(APIClient):classConfig:base_url='http://things.depop.com'get_things=GetEndpoint('/things/{user_id}/',# url (format string)querystring_args=(Arg('since',default=now),),request_headers={# added to all requests'Authorization':'Bearer {access_token}'# (format string)})response_data=ThingServiceClient.get_things(user_id=123)# GET http://things.depop.com/things/123/?since=2018-02-09T13:31:10.569481
您仍然可以在每次调用时向requestslib传递额外的参数 使用_request_kwargs:
response_data=ThingServiceClient.get_things(user_id=123,type='cat',offset_id='65345ff34e344ab53c',limit=20,access_token='87a64c98b62d39e8625f',_request_kwargs={'headers':{'X-Depop-WTF':'something something'}},)
对于带有请求正文的呼叫:
frompopgetimportAPIClient,PostEndpoint,BodyTypeclassThingServiceClient(APIClient):classConfig:base_url='http://things.depop.com'new_thing=PostEndpoint('/things/',body_required=True,body_type=BodyType.FORM_ENCODED,# or BodyType.JSON - sets content-type headerrequest_headers={'Authorization':'Bearer {access_token}',})response_data=ThingServiceClient.new_thing(access_token='87a64c98b62d39e8625f',body={'type':'dog','name':'fido',})
您还可以使用_sessionkwarg:
fromdjango.confimportsettingsfromrequests_oauthlibimportOAuth1Sessionfrommyproject.twitter.clientimportTwitterClientdeftweet_from(user,message,**extra):oauth_session=OAuth1Session(settings.TWITTER_CONSUMER_KEY,client_secret=settings.TWITTER_CONSUMER_SECRET,resource_owner_key=user.tw_access_token,resource_owner_secret=user.tw_access_token_secret,)body={'status':message,}body.update(extra)returnTwitterClient.update_status(body=body,_session=oauth_session)
异步
通过 requests-futures后端。
pip install popget[threadpool]
提供了APIClient的异步变量,它将同时具有异步和阻塞 所有终结点方法的版本。
frompopgetimportArg,GetEndpointfrompopget.async.threadpoolimportAPIClientimportrequestsclassThingServiceClient(APIClient):classConfig:base_url='http://things.depop.com'get_things=GetEndpoint('/things/{user_id}/',# url format stringquerystring_args=(Arg('type',required=True),# required querystring param (validated on call)),)# blocking:data=ThingServiceClient.get_things(user_id=2345,type='cat')# async:future=ThingServiceClient.async_get_things(user_id=2345,type='cat')# response is parsed and may raise, as for blocking requeststry:data=future.result()exceptrequests.exceptions.HTTPErrorase:print(repr(e))
异步终结点方法将返回标准的concurrent.futures.Future对象。
您可以自定义生成的异步终结点方法的名称:
classThingServiceClient(APIClient):classConfig:base_url='http://things.depop.com'async_method_template='{}__async'get_things=GetEndpoint('/things/{user_id}/',# url format stringquerystring_args=(Arg('type',required=True),# required querystring param (validated on call)),)future=ThingServiceClient.get_things__async(user_id=2345,type='cat')
兼容性
此项目的测试依据是:
Python 2.7 | |
Python 3.7 |
运行测试
圆圈
cli目前不支持“工作流”,因此必须运行 两个python版本的作业分别是:
circleci build --job python-2.7
circleci build --job python-3.7
py.test(单一python版本)
也可以在本地运行测试,允许调试 发生的错误。
决定要测试哪个python版本并创建一个virtualenv:
pyenv virtualenv 3.7.4 popget
pip install -r requirements-test.txt
py.test -v -s --ipdb tests/