项目描述
1、为客制化的Web API接口建立可重用的对象
2、可以与Requests,asyncio和Twisted
快速开始
Uplink把你的HTTP API接口转化成一个Python的类
from uplink import Consumer, get, headers, Path, Query class GitHub(Consumer): @get("users/{user}/repos") def get_repos(self, user: Path, sort_by: Query("sort")): """获取到用户的公共存储库资源"""
建立一个可以与webservice交互的实例:
github = GitHub(base_url="https://api.github.com/")
base_url是整个uplink架构的基础,指向需要对接的API的域名,比如我是在写django项目,可以直接将这个参数集成到settings文件内,包括可以建立字典参数可选化等等。
现在,执行一个HTTP请求已经简单的变成调用一个方法。
repos = github.get_repos("octocat", sort_by="created")
前面一个参数"octocat",对应的就是类里对应方法的get(也可能是post或者其他装饰上)的user参数,后面的就不必多说,sort_by的传参。这里形成的新url其实就是
https://api.github.com/users/octocat/repos/?sort=created
返回的是一个友好的requests.Response对象
print(repos.json()) # Output: [{'id': 64778136, 'name': 'linguist', ...
这里的.json()实际上就是一个反序列化的过程。
如果想发送一个非阻塞的请求,uplink支持使用aiohttp和twisted。
使用装饰器和功能参数注解来描述HTTP request:
1、URL参数的替代和查询参数的支持
2、将响应体转换成Python的对象(等等,使用marshmallow或者一个客制化的转换)
3、JSON、URL编码和多重庆球体和文件上传
4、注入功能像中间件来使用客制化的返回和错误处理
安装
pip3 install uplink
静态请求处理
方法装饰器描述与所有Consumer方法的所有调用相关的请求属性。
例如下面的GitHub API consumer:
class GitHub(uplink.Consumer): @uplink.timeout(60) @uplink.get("/repositories") def get_repos(self):
调用了timeout函数,get_repos方法将会构建HTTP请求然后等带60秒,如果服务器还没有在这个时间之前没有返回就会放弃。由于方法注释仅仅是装饰器,您可以堆叠一个在另一个之上:
class GitHub(uplink.Consumer): @uplink.headers({"Accept": "application/vnd.github.v3.full+json"}) @uplink.timeout(60) @uplink.get("/repositories") def get_repos(self):
动态请求处理
对于一个常规的项目而言,方法参数驱动着一个方法的动态行为;一个方法的输出通常依赖于它的输入。在uplink里面,函数参数使HTTP请求参数化,你可以标识请求的动态部分通过适当的申明这些参数。为了说明,在下面代码中的get_user()方法,我们已经申明了参数username作为URL占位符的替代, 使用Path来注释
class GitHub(uplink.Consumer): @uplink.get("users/{username}") def get_user(self, username: uplink.Path("username")): pass
通过consumer实例来调用这个方法
github.get_user(username="prkumar")
创建的HTTP请求的url是以users/prkumar结尾的。
请求方法
uplink提供装饰器将任何方法转化到请求的定义中。这些装饰器提供请求方法和相关的url包含get、post、put、patch、和delete
相关的url资源被规定在装饰器内部
@get("users/list")
你也可以定义查询参数在url里面
@get("users/list?sort=desc")
另外请求方法必须绑定到Consumer的子类
class MyApi(Consumer): @get("users/list") def list_users(self)
操纵URL
一个请求的url使用url模板参数可以被动态的更新。一个简单的URL参数是一个包含数字和字母的的字符串被{}两个花括号包围。
若要讲参数与方法的参数匹配,则将参数的名称与字母数字字符串匹配,如下:
@get("group/{id}/users") def group_list(self, id): pass
或者使用Path注释
@get("group/{id}/users") def group_list(self, group_id: Path("id")): pass
也可以添加Query参数
@get("group/{id}/users") def group_list(self, group_id: Path("id"), sort: Query): pass
对于复杂的查询参数组合,可以使用映射
@get("group/{id}/users") def group_list(self, group_id: Path("id"), options: QueryMap): pass
请求体
可以指定一个参数的值作为HTTP请求体,并用Body标注:
@post("users/new") def create_user(self, user: Body): pass
这个注释与关键字参数(由**前缀表示)可以很好地工作:
@post("users/new") def create_user(self, **user_info: Body): pass
表单编码,Multipart和JSON
方法也可以被申明来发送表单编码, multipart和json数据
表单编码数据在装饰form_url_encoded方法的时候被发送,每一对键值对都使用Field进行注解:
@form_url_encoded @post("user/edit") def update_user(self, first_name: Field, last_name: Field): pass
Multipart 在被multipart函数装饰的时候使用,内容的申明使用Part注解
@multipart @put("user/photo") def update_user(self, photo: Part, description: Part): pass
JSON数据同理在被json函数装饰的时候发送。Body注解声明json的有效载荷
@uplink.json @uplink.patch("/user") def update_user(self, **user_info: uplink.Body):
操纵头部
可以使用headers装饰器发送静态的请求头信息
@headers({ "Accept": "application/vnd.github.v3.full+json", "User-Agent": "Uplink-Sample-App" }) @get("users/{username}") def get_user(self, username): pass
headers也可以被用来当做类的装饰器,当每一个请求都需要被添加这个装饰的时候
@headers({ "Accept": "application/vnd.github.v3.full+json", "User-Agent": "Uplink-Sample-App" }) class GitHub(Consumer): ....
请求头可以被动态更新当使用Header方法参数注解
@get("user") def get_user(self, authorization: Header):
反序列化响应体
uplink让转换Httpp响应体到数据模型对象变得简单并且是可选择的,无论是使用uplink内置库的函数支持比如:marshmallow(可以查看uplink.converters.MarshmallowConverter
)或者使用uplink.loads来写客制化的转换逻辑满足自己独特的要求。
最基础的,你需要使用装饰器uplink.returns模块标注想要返回的类型.而当你着手与JSON响应的API的时候uplink.json可以很好的完成。
@returns.json(User) @get("users/{username}") def get_user(self, username): pass
Python3的用户另外还可以使用返回类型的暗示
@returns.json @get("users/{username}") def get_user(self, username) -> User: pass
最后的步骤是注册一个将HTTP响应转换成预期返回类型的策略。因此,uplink.loads()可以注册一个函数,该函数可以处理特定类及其子类的反序列化。
from models import ModelBase # 所有模型类型的基类,包含上面的User # 告诉uplink如何将JSON响应反序列化成我们的模型类 @loads.install #使这个对所有consumer实例有效 @loads.from_json(ModelBase) def load_model_from_json(model_cls, json_obj): return model_cls.from_json(json_obj)
这一步不是必须的如果你定义你的数据模型对象使用uplink的内置库支持比如marshmallow
订制响应跟错误处理
想要订制响应跟错误处理,需要装饰response_handler或者error_handler装饰器。
例如,方法returns_success()定义如下,它是一个响应处理输出,无论请求是否成功。
@uplink.response_handler def returns_success(response): return response.status == 200
现在returns_success()可以被用来像装饰器一样使用,来注入客制化的请求处理到任何请求方法中。
@returns_success @put("/todos") def create_todo(self, title):
为了使其作用于所有的请求方法的Consumer子类中,也可以简单的使用注册处理的方法使其作为一个类装饰器
@returns_success class TodoApp(uplink.Consumer): ...
相似的,error_handler被注册来装饰方法。当需要使用这个请求方法的时候调用它。应用此请求方法,当底层HTTP客户端无法执行请求时调用这些处理程序:
@error_handler def raise_api_error(exc_type, exc_val, exc_tb): # wrap client error with custom API error ...
值得注意的是,装饰器可以置于另一个之上链接他们的行为
@raise_api_error @returns_success class TodoApp(uplink.Consumer): ...
关于__init__()
例如下面的consumer接受API访问token作为结构化参数的access_token:
class GitHub(uplink.Consumer): def __init__(self, access_token: uplink.Query): ... @uplink.post("/user") def update_user(self, **info: Body): """Update the authenticated user"""
现在,所有这个consumer类生成的实例请求将会在初始化的时候被access token认证
github = TodoApp("my-github-access-token") github.update_user(bio="Beam me up, Scotty!")
_inject()请求属性
作为一个可选择的__init__的参数,你可以获得相似的行文和更多的控制通过使用Consumer._inject()方法. 通过这个方法,你可以计算请求的属性在普通的python方法中。
class TodoApp(uplink.Consumer): def __init__(self, username, password) # Create an access token api_key = create_api_key(username, password) # Inject it. self._inject(uplink.Query("api_key").with_value(api_key))
跟之前的注解风格相似,被_inject()添加的请求属性可以作用于全部的consumer实例的请求。
最后看一个整体实例:
此实例来自于一个django项目,url配置于django的settings文件中
看一下调用