未提供项目说明
lc-api-manager的Python项目详细描述
api管理器py
概述
许多具有速率限制的API将管理速率限制的责任推给API用户,包括那些具有sdk的API。
api-manager-py
是一个Python库,旨在抽象出管理速率限制的复杂性,
允许开发人员专注于检索所需的数据,而无需拼凑出有缺陷的、逐案解决方案。
API管理功能不需要使用特定的库来与API交互,并且
根据输入参数自动缓存响应以减少网络IO。只需实现一个小型API客户机
接口并开始提出请求。在
安装
从PyPi安装(首选方法)
pip install lc-api-manager
使用Pip
从GitHub安装 ^{pr2}$其中x.x.x
是要下载的版本。在
手动下载安装
要下载源分发和/或控制盘文件,请导航到
https://github.com/libcommon/api-manager-py/tree/releases/vx.x.x/dist
,其中x.x.x
是要安装的版本,
并通过用户界面或wget等工具进行下载。然后安装运行:
pip install <downloaded file>
下载后请不要更改文件名,因为Pip需要安装文件的特定命名约定。在
依赖关系
api-manager-py
依赖于用于缓存API响应的lc-cache库。仅限
正式支持Python>;=3.6版本。在
入门
第一步是实现APIClient
,并选择一个库来发出HTTP请求。一个常见的选择是
Requests库,我们将使用它来实现GitHub REST API
v3的客户机。GitHub的API的域是https://api.github.com
,因此客户机的request
方法
只需要HTTP方法(GET
,POST
,等等)、API端点(即/repos/<username>/<repo_name>
)和可选的headers
,
params
,和data
字典(参见:Requests documentation)。在
fromhashlibimportsha256fromtypingimportAny,Dict,OptionalimportrequestsfromrequestsimportResponsefromlc_api_managerimportAPIClientclassGitHubAPIClient(APIClient):"""API client for GitHub Rest API v3."""__slots__=("_headers",)def__init__(self,auth_token:Optional[str]=None)->None:"""Initialize API client with optional GitHub API oauth token see: https://developer.github.com/v3/#authentication. """ifauth_token:self._headers={"Authorization":"token {}".format(auth_token)}else:self._headers=dict()defprocess_response_for_cache(self,response:Optional[Response])->Optional[str]:"""Return the SHA-256 hash of the API response if not None."""ifresponse:returnsha256(response.text.encode("utf8")).hexdigest()returnNonedefrequest(http_method:str,api_endpoint:str,headers:Optional[Dict[str,Any]],params:Optional[Dict[str,Any]],data:Optional[Dict[Any,Any]])->Response:"""Make request to GitHub REST API endpoint with provided headers, URL parameters, and data and return response."""# Merge authorization header with provided headersmerged_headers=self._headersifheaders:merged_headers.update(headers)# Construct full URL and make requesturl="https://api.github.com/{}".format(api_endpoint.lstrip("/"))response=requests.request(http_method.upper(),url,params=params,data=data,headers=merged_headers)# Raise error if status code is 4XX or 5XXresponse.raise_for_status()returnresponse
有了函数APIClient
,我们可以使用内置的APIManager
类发出请求并缓存它们:
fromlc_api_managerimportAPIManagerfromlc_cacheimportHashmapCachedefmain()->int:"""Make 60 unauthenticated requests to an API endpoint in rapid succession."""api_manager=APIManager(3600,# GitHub API allows 60 unauthenticated requests per hour60,GitHubAPIClient(),HashmapCache())# Make 60 requests to the same API endpointfor_inrange(60):# The API manager will make the request on first iteration,# but will return cached response on the other 59response=api_manager.request("GET","repos/libcommon/api-manager-py",params=dict(per_page=100))# Check rate limit status, should be 59 requests remaining# See: https://developer.github.com/v3/rate_limit/response=api_manager.request("GET","rate_limit")requests_remaining=response.json().get("resources").get("core").get("remaining")assert(requests_remaining==59)return0if__name__=="__main__":main()
如果您正在运行多个Python进程从同一API请求数据,并且希望确保所有这些进程都
根据速率限制要求,可以重写APIManager.update_state
方法。APIManager
构造函数
有一个名为updated_state_before_request
的参数,默认为False。如果将其设置为True
,则update_state
方法将在每个API请求之前调用,因此可用于跨多个进程同步速率限制状态。
例如,可以使用/rate_limit
GitHub API端点来实现此方法:
fromlc_api_managerimportAPIManagerdefGitHubAPIManager(APIManager):"""API manager for GitHub REST API v3 that syncs rate limiting state."""__slots__=()def__init__(self,*args,**kwargs)->None:ifnotkwargs.get("update_state_before_request"):kwargs["update_state_before_request"]=Truesuper().__init__(*args,**kwargs)defupdate_state(self)->None:"""Make request to /rate_limit endpoint and update rate limit status."""response=self._client.request("GET","rate_limit")requests_remaining=response.json().get("resources").get("core").get("remaining")self._count=self._threshold-requests_remaining
贡献/建议
欢迎投稿和建议!要提出功能请求,请报告错误,或对现有的 功能,请提交问题。如需投稿,请提交一份个人简历,但请务必进行打印、类型检查和测试 你的代码。提前谢谢!在
- 项目
标签: