Python 转路由之uplink

提示：扫一扫查出行
【扫一扫了解最新限行尾号】
复制提示

项目描述

1、为客制化的Web API接口建立可重用的对象

2、可以与Requests，asyncio和Twisted

快速开始

Uplink把你的HTTP API接口转化成一个Python的类

from uplink import Consumer, get, headers, Path, Queryclass 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_handlerdef returns_success(response): return response.status == 200

现在returns_success()可以被用来像装饰器一样使用，来注入客制化的请求处理到任何请求方法中。

@returns_success@put("/todos")def create_todo(self, title):

为了使其作用于所有的请求方法的Consumer子类中，也可以简单的使用注册处理的方法使其作为一个类装饰器

@returns_successclass TodoApp(uplink.Consumer): ...

相似的，error_handler被注册来装饰方法。当需要使用这个请求方法的时候调用它。应用此请求方法，当底层HTTP客户端无法执行请求时调用这些处理程序：

@error_handlerdef raise_api_error(exc_type, exc_val, exc_tb): # wrap client error with custom API error ...

值得注意的是，装饰器可以置于另一个之上链接他们的行为

@raise_api_error@returns_successclass TodoApp(uplink.Consumer): ...

关于init()

函数注释（如Query和Header）可以与consumer子类的构造函数参数一起使用。当创建新的consumer实例时，这些参数的值应用于通过该实例所做的所有请求。

例如下面的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实例的请求。

最后看一个整体实例：