banner
李大仁博客

李大仁博客

天地虽大,但有一念向善,心存良知,虽凡夫俗子,皆可为圣贤。

omi_async_http_client

omi_async_http_client#

在 github.com 上 Fork omi_async_http_client https://github.com/limccn/omi_async_http_client

描述#

使用 asyncio 和后端实现的异步 http 客户端

用法#

  1. 从 pip 安装 omi_async_http_client

$pip install omi_async_http_client

或者从源代码安装

$python setup.py install

  1. 为 omi_async_http_client 安装后端,支持同步和异步

使用异步后端,例如aiohttphttpx

$pip install aiohttp

$pip install httpx

或者使用传统的同步后端requests

$pip install requests

  1. 应用到你的项目中。

从 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

  1. 测试 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
)

  1. 我们实现了一个使用FastAPI的演示 API 提供程序,展示了如何使用这个库,并包含了测试。

@See mock_fastapi.py 了解详情

加载中...
此页面数据所有权由区块链加密技术和智能合约保障仅归创作者所有。