snug 2.3.1

为什么？

编写可重用的 Web API 交互很困难。考虑一个通用示例：

import json

def repo(name, owner):
    """get a github repo by owner and name"""
    request = Request(f'https://api.github.com/repos/{owner}/{name}')
    response = my_http_client.send(request)
    return json.loads(response.content)

很好很简单。但…

• 异步呢？我们是否为此编写另一个函数？

• 我们如何为此编写干净的单元测试？

• 如果我们想使用另一个 HTTP 客户端或会话怎么办？

• 我们如何将其与不同的凭据一起使用？

Snug允许您编写独立于 HTTP 客户端、凭据或它们是否（a）同步运行的 API 交互。

与大多数 API 客户端工具包相比，snug 为您做出最少的假设和设计决策。其简单、适应性强的基础确保您可以专注于使您的 API 独一无二的原因。无论您是在编写功能齐全的 API 包装器，还是只是进行一些 API 调用，Snug 都非常适合。

快速开始

1. API 交互（“查询”）是请求/响应生成器。

import snug

def repo(name, owner):
    """get a github repo by owner and name"""
    request = snug.GET(f'https://api.github.com/repos/{owner}/{name}')
    response = yield request
    return json.loads(response.content)

2. 可以执行查询：

>>> query = repo('Hello-World', owner='octocat')
>>> snug.execute(query)
{"description": "My first repository on Github!", ...}

特征

1. 毫不费力地异步。同样的查询也可以异步执行：

   query = repo('Hello-World', owner='octocat')
   repo = await snug.execute_async(query)

2. 灵活性。由于查询只是生成器，因此自定义它们不需要特殊的胶水代码。例如：添加验证逻辑，或者使用任何序列化方法：

   from my_types import User, UserSchema
   
   def user(name: str) -> snug.Query[User]:
       """lookup a user by their username"""
       if len(name) == 0:
           raise ValueError('username must have >0 characters')
       request = snug.GET(f'https://api.github.com/users/{name}')
       response = yield request
       return UserSchema().load(json.loads(response.content))

3. 可插拔客户端。查询与 HTTP 客户端完全无关。例如，要使用请求 而不是标准库：

   import requests
   query = repo('Hello-World', owner='octocat')
   snug.execute(query, client=requests.Session())

   阅读这里 如何注册您自己的。

4. 可测试性。无需接触网络即可轻松运行查询。不需要复杂的模拟或猴子补丁。

   >>> query = repo('Hello-World', owner='octocat')
   >>> next(query).url.endswith('/repos/octocat/Hello-World')
   True
   >>> query.send(snug.Response(200, b'...'))
   StopIteration({"description": "My first repository on Github!", ...})

5. 可交换身份验证。查询与会话或凭据无关。使用不同的凭据来执行相同的查询：

   def follow(name: str) -> snug.Query[bool]:
       """follow another user"""
       req = snug.PUT('https://api.github.com/user/following/{name}')
       return (yield req).status_code == 204
   
   snug.execute(follow('octocat'), auth=('me', 'password'))
   snug.execute(follow('octocat'), auth=('bob', 'hunter2'))

6. 相关查询。使用基于类的查询为相关对象创建富有表现力的链式 API：

   class repo(snug.Query[dict]):
       """a repo lookup by owner and name"""
       def __init__(self, name, owner): ...
   
       def __iter__(self): ...  # query for the repo itself
   
       def issue(self, num: int) -> snug.Query[dict]:
           """retrieve an issue in this repository by its number"""
           r = snug.GET(f'/repos/{self.owner}/{self.name}/issues/{num}')
           return json.loads((yield r).content)
   
   my_issue = repo('Hello-World', owner='octocat').issue(348)
   snug.execute(my_issue)

7. 分页。为（异步）迭代定义分页查询。

   def organizations(since: int=None):
       """retrieve a page of organizations since a particular id"""
       resp = yield snug.GET('https://api.github.com/organizations',
                             params={'since': since} if since else {})
       orgs = json.loads(resp.content)
       next_query = organizations(since=orgs[-1]['id'])
       return snug.Page(orgs, next_query=next_query)
   
   my_query = snug.paginated(organizations())
   
   for orgs in snug.execute(my_query):
       ...
   
   # or, with async
   async for orgs in snug.execute_async(my_query):
       ...

8. 基于函数还是基于类？你决定。保持一切 DRY 的一种选择是使用基于类的查询和继承：

   class BaseQuery(snug.Query):
       """base github query"""
   
       def prepare(self, request): ...  # add url prefix, headers, etc.
   
       def __iter__(self):
           """the base query routine"""
           request = self.prepare(self.request)
           return self.load(self.check_response((yield request)))
   
       def check_response(self, result): ...  # raise nice errors
   
   class repo(BaseQuery):
       """get a repo by owner and name"""
       def __init__(self, name, owner):
           self.request = snug.GET(f'/repos/{owner}/{name}')
   
       def load(self, response):
           return my_repo_loader(response.content)
   
   class follow(BaseQuery):
       """follow another user"""
       def __init__(self, name):
           self.request = snug.PUT(f'/user/following/{name}')
   
       def load(self, response):
           return response.status_code == 204

   或者，如果您对高阶函数和装饰器感到满意，请使用gentools 修改查询yield、send和返回值：

   from gentools import (map_return, map_yield, map_send,
                         compose, oneyield)
   
   class Repository: ...
   
   def my_repo_loader(...): ...
   
   def my_error_checker(...): ...
   
   def my_request_preparer(...): ...  # add url prefix, headers, etc.
   
   basic_interaction = compose(map_send(my_error_checker),
                               map_yield(my_request_preparer))
   
   @map_return(my_repo_loader)
   @basic_interaction
   @oneyield
   def repo(owner: str, name: str) -> snug.Query[Repository]:
       """get a repo by owner and name"""
       return snug.GET(f'/repos/{owner}/{name}')
   
   @basic_interaction
   def follow(name: str) -> snug.Query[bool]:
       """follow another user"""
       response = yield snug.PUT(f'/user/following/{name}')
       return response.status_code == 204

有关更多信息，请查看教程、 高级功能、 食谱或示例。

安装

没有必需的依赖项。安装很简单，因为：

pip install snug

虽然 snug 包括基本的同步和异步 HTTP 客户端，但您可能希望安装requests、 httpx和/或aiohttp。

pip install requests aiohttp httpx

备择方案

如果您正在寻找不那么简约的 API 客户端工具包，请查看uplink 或tapioca。