omi_async_http_client#
在 github.com 上 Fork omi_async_http_client https://github.com/limccn/omi_async_http_client
描述#
使用 asyncio 和后端实现的异步 http 客户端
用法#
- 从 pip 安装 omi_async_http_client
$pip install omi_async_http_client
或者从源代码安装
$python setup.py install
- 为 omi_async_http_client 安装后端,支持同步和异步
$pip install aiohttp
$pip install httpx
或者使用传统的同步后端requests
$pip install requests
- 应用到你的项目中。
从 omi_async_http_client.APIClient 设置一个 TEMPLATE APIClient 构建函数,
当使用你的 TEMPLATE APIClient 时,omi_async_http_client 会自动填充后端参数以执行 http 请求。
from omi_async_http_client import APIClient as APIClientBuilder
from app.config import settings
def my_api_client_builder(model):
return APIClientBuilder(
model=model, # None is OK
http_backend="omi_async_http_client.AioHttpClientBackend", # 选择 aiohttp 作为后端
resource_endpoint=settings.get("API_ENDPOINT_URL", ""),
client_id=settings.get("API_ENDPOINT_CLIENT_ID", ""),
client_secret=settings.get("API_ENDPOINT_CLIENT_SECRET", "")
)
MyAPIClient = my_api_client_builder
为请求定义一个用户模型,使用@RequestModel
装饰器来定义你的 API,别名@api_request_model
在使用@RequestModel
时起到同样的作用。
from pydantic import BaseModel
from typing import List
from omi_async_http_client import RequestModel
@RequestModel(api_name="/staff/{id}", api_prefix="", api_suffix="")
class Staff(BaseModel):
id
name = ""
age = 0
gender = "F"
class PagedStaff(BaseModel):
page
offset
limit
staffsStaff
- 测试 HTTP 客户端是否工作,并享受 omi_async_http_client。
client = APIClient (Staff) # 使用 Staff 创建一个 Apiclient
从 Restful API 中检索一些数据#
response = await client.retrieve(
condition={ # 用于分页的额外参数
'id':123, # 将填充 {id} 占位符,将 /staff/{id} 更改为 /staff/123。
'name': 'python'
},
extra_params={ # 用于分页的额外参数
'page': 1,
'offset': 1,
'limit': 10
},
paging_model=PagedStaff
)
- 我们实现了一个使用FastAPI的演示 API 提供程序,展示了如何使用这个库,并包含了测试。
@See mock_fastapi.py 了解详情