优雅地管理您的API交互
项目描述
Python最优雅的API客户端框架
Gracy处理所有HTTP交互的失败、日志记录、重试、节流、解析和报告。Gracy在底层使用httpx。
"让Gracy做无聊的事情,您则专注于您的应用"
摘要
🧑💻 开始使用
安装
pip install gracy
或
poetry add gracy
使用
示例将使用 PokeAPI 进行展示。
简单示例
# 0. Import
import asyncio
import typing as t
from gracy import BaseEndpoint, Gracy, GracyConfig, LogEvent, LogLevel
# 1. Define your endpoints
class PokeApiEndpoint(BaseEndpoint):
GET_POKEMON = "/pokemon/{NAME}" # 👈 Put placeholders as needed
# 2. Define your Graceful API
class GracefulPokeAPI(Gracy[str]):
class Config:
BASE_URL = "https://pokeapi.co/api/v2/" # 👈 Optional BASE_URL
# 👇 Define settings to apply for every request
SETTINGS = GracyConfig(
log_request=LogEvent(LogLevel.DEBUG),
log_response=LogEvent(LogLevel.INFO, "{URL} took {ELAPSED}"),
parser={
"default": lambda r: r.json()
}
)
async def get_pokemon(self, name: str) -> t.Awaitable[dict]:
return await self.get(PokeApiEndpoint.GET_POKEMON, {"NAME": name})
pokeapi = GracefulPokeAPI()
async def main():
try:
pokemon = await pokeapi.get_pokemon("pikachu")
print(pokemon)
finally:
pokeapi.report_status("rich")
asyncio.run(main())
更多示例
设置
严格/允许的状态码
默认情况下,Gracy 将任何成功的状态码(200-299)视为成功。
严格
您可以通过定义严格状态码或增加允许的状态码范围来修改此行为
from http import HTTPStatus
GracyConfig(
strict_status_code=HTTPStatus.CREATED
)
或一个值列表
from http import HTTPStatus
GracyConfig(
strict_status_code={HTTPStatus.OK, HTTPStatus.CREATED}
)
使用 strict_status_code
意味着任何未指定的其他代码将引发错误,无论是否成功。
允许
您也可以保持该行为,但扩展允许的代码范围。
from http import HTTPStatus
GracyConfig(
allowed_status_code=HTTPStatus.NOT_FOUND
)
或一个值列表
from http import HTTPStatus
GracyConfig(
allowed_status_code={HTTPStatus.NOT_FOUND, HTTPStatus.FORBIDDEN}
)
使用 allowed_status_code
意味着所有成功代码加上您定义的代码将被视为成功。
这很快就会在解析中看到。
⚠️ 注意,strict_status_code
优先于 allowed_status_code
,您可能不想将它们组合在一起。请选择一个或另一个。
自定义验证器
您可以实现自己的自定义验证器来进一步检查响应并决定是否将请求视为失败(从而触发重试,如果已设置)。
from gracy import GracefulValidator
class MyException(Exception):
pass
class MyCustomValidator(GracefulValidator):
def check(self, response: httpx.Response) -> None:
jsonified = response.json()
if jsonified.get('error', None):
raise MyException("Error is not expected")
return None
...
class Config:
SETTINGS = GracyConfig(
...,
retry=GracefulRetry(retry_on=MyException, ...), # Set up retry to work whenever our validator fails
validators=MyCustomValidator(), # Set up validator
)
解析
解析允许您根据返回的状态码处理请求。
基本示例是解析 json
GracyConfig(
parser={
"default": lambda r: r.json()
}
)
在此示例中,所有成功的请求将自动返回 json()
结果。
您还可以将其缩小到处理特定状态码。
class Config:
SETTINGS = GracyConfig(
...,
allowed_status_code=HTTPStatusCode.NOT_FOUND,
parser={
"default": lambda r: r.json()
HTTPStatusCode.NOT_FOUND: None
}
)
async def get_pokemon(self, name: str) -> dict| None:
# 👇 Returns either dict or None
return await self.get(PokeApiEndpoint.GET_POKEMON, {"NAME": name})
甚至可以自定义 异常以改进代码可读性
class PokemonNotFound(GracyUserDefinedException):
... # More on exceptions below
class Config:
GracyConfig(
...,
allowed_status_code=HTTPStatusCode.NOT_FOUND,
parser={
"default": lambda r: r.json()
HTTPStatusCode.NOT_FOUND: PokemonNotFound
}
)
async def get_pokemon(self, name: str) -> Awaitable[dict]:
# 👇 Returns either dict or raises PokemonNotFound
return await self.get(PokeApiEndpoint.GET_POKEMON, {"NAME": name})
解析类型
因为解析器允许您根据状态码动态解析有效负载,所以您的 IDE 不会自动识别返回类型。
为了避免在每个方法上使用无聊的 typing.cast
,Gracy 提供了类型化的 HTTP 方法,因此您可以定义特定的返回类型
async def list(self, offset: int = 0, limit: int = 20):
params = dict(offset=offset, limit=limit)
return await self.get[ResourceList]( # Specifies this method return a `ResourceList`
PokeApiEndpoint.BERRY_LIST, params=params
)
async def get_one(self, name_or_id: str | int):
return await self.get[models.Berry | None](
PokeApiEndpoint.BERRY_GET, params=dict(KEY=str(name_or_id))
)
重试
谁不喜欢不可靠的 API? 🙋
然而,这样的 API 很多。
使用 tenacity、backoff、retry、aiohttp_retry 以及其他任何重试库并不容易。 🙅
您仍然需要为每个请求编码实现,这很烦人。
这就是 Gracy 允许您实现重试逻辑的方式
class Config:
GracyConfig(
retry=GracefulRetry(
delay=1,
max_attempts=3,
delay_modifier=1.5,
retry_on=None,
log_before=None,
log_after=LogEvent(LogLevel.WARNING),
log_exhausted=LogEvent(LogLevel.CRITICAL),
behavior="break",
)
)
参数 | 描述 | 示例 |
---|---|---|
delay |
两次重试之间的等待秒数 | 2 将等待 2 秒,1.5 将等待 1.5 秒,依此类推 |
max_attempts |
Gracy 应该重试请求多少次? | 10 表示 1 次常规请求,另外 10 次重试(如果它们继续失败)。 1 应该是最小值 |
delay_modifier |
允许您通过将此值乘以 delay 来指定递增延迟时间 |
设置 1 表示没有延迟更改。设置 2 表示每次重试时延迟将加倍 |
retry_on |
我们应该为哪些状态码/异常重试? None 表示任何非成功状态码或异常 |
HTTPStatus.BAD_REQUEST 或 {HTTPStatus.BAD_REQUEST, HTTPStatus.FORBIDDEN} 或 Exception 或 {Exception, HTTPStatus.NOT_FOUND} |
log_before |
指定日志级别。 None 表示不记录 |
关于日志记录的更多信息稍后提供 |
log_after |
指定日志级别。 None 表示不记录 |
关于日志记录的更多信息稍后提供 |
log_exhausted |
指定日志级别。 None 表示不记录 |
关于日志记录的更多信息稍后提供 |
行为 |
允许您定义在重试失败时如何处理。 pass 将接受任何重试失败 |
pass 或 break (默认值) |
overrides |
允许根据最后响应的状态码覆盖 delay |
{HTTPStatus.BAD_REQUEST: OverrideRetryOn(delay=0), HTTPStatus.INTERNAL_SERVER_ERROR: OverrideRetryOn(delay=10)} |
节流
速率限制问题?不再有了。
Gracy 帮助您在 API 向您抛出 429 之前主动处理它。
创建规则
您可以使用正则表达式为每个端点定义规则。
SIMPLE_RULE = ThrottleRule(
url_pattern=r".*",
max_requests=2
)
print(SIMPLE_RULE)
# Output: "2 requests per second for URLs matching re.compile('.*')"
COMPLEX_RULE = ThrottleRule(
url_pattern=r".*\/pokemon\/.*",
max_requests=10,
per_time=timedelta(minutes=1, seconds=30),
)
print(COMPLEX_RULE)
# Output: 10 requests per 90 seconds for URLs matching re.compile('.*\\/pokemon\\/.*')
设置节流。
您可以为日志设置规则并分配。
class Config:
GracyConfig(
throttling=GracefulThrottle(
rules=ThrottleRule(r".*", 2), # 2 reqs/s for any endpoint
log_limit_reached=LogEvent(LogLevel.ERROR),
log_wait_over=LogEvent(LogLevel.WARNING),
),
)
并发请求
也许您正在调用的API有一些慢速端点,您想确保并发请求的数量不超过自定义数量。
您可以定义一个ConcurrentRequestLimit
配置。
最简单的用法是
from gracy import ConcurrentRequestLimit
class Config:
GracyConfig(
concurrent_requests=ConcurrentRequestLimit(
limit=1, # How many concurrent requests
log_limit_reached=LogEvent(LogLevel.WARNING),
log_limit_freed=LogEvent(LogLevel.INFO),
),
)
但您也可以按方法轻松定义它
class MyApiClient(Gracy[Endpoint]):
@graceful(concurrent_requests=5)
async def get_concurrently_five(self, name: str):
...
日志记录
您可以使用LogEvent
和LogLevel
来定义和自定义日志事件。
verbose_log = LogEvent(LogLevel.CRITICAL)
custom_warn_log = LogEvent(LogLevel.WARNING, custom_message="{METHOD} {URL} is quite slow and flaky")
custom_error_log = LogEvent(LogLevel.INFO, custom_message="{URL} returned a bad status code {STATUS}, but that's fine")
请注意,占位符将被Gracy根据事件类型格式化和替换,例如
每个事件占位符
占位符 | 描述 | 示例 | 支持的事件 |
---|---|---|---|
{URL} |
目标的全url | https://pokeapi.co/api/v2/pokemon/pikachu |
所有 |
{UURL} |
目标的全未格式化url | https://pokeapi.co/api/v2/pokemon/{NAME} |
所有 |
{ENDPOINT} |
目标端点 | /pokemon/pikachu |
所有 |
{UENDPOINT} |
未格式化端点 | /pokemon/{NAME} |
所有 |
{METHOD} |
使用的HTTP请求方法 | GET , POST |
所有 |
{STATUS} |
响应返回的状态码 | 200 , 404 , 501 |
请求后 |
{ELAPSED} |
请求完成所需的时间(秒) | 数字 | 请求后 |
{REPLAY} |
仅在请求重放时显示的占位符 | REPLAYED 当重放时,否则为空字符串(``) |
请求后 |
{IS_REPLAY} |
表示是否重放的布尔值 | 当重放时为字符串TRUE ,否则为FALSE |
请求后 |
{RETRY_DELAY} |
Gracy在重复请求之前等待的时间 | 数字 | 任何重试事件 |
{RETRY_CAUSE} |
触发重试逻辑的原因 | [Bad Status Code: 404] ,[Request Error: ConnectionTimeout] |
任何重试事件 |
{CUR_ATTEMPT} |
当前请求的当前尝试次数 | 数字 | 任何重试事件 |
{MAX_ATTEMPT} |
为当前请求定义的最大尝试次数 | 数字 | 任何重试事件 |
{THROTTLE_LIMIT} |
为当前请求定义的请求数量/秒 | 数字 | 任何节流事件 |
{THROTTLE_TIME} |
Gracy在调用请求之前等待的时间 | 数字 | 任何节流事件 |
{THROTTLE_TIME_RANGE} |
节流规则定义的时间范围 | second ,90 seconds |
任何节流事件 |
您可以根据以下方式设置日志事件
请求
- 请求前
- 响应后
- 响应有非成功的错误
GracyConfig(
log_request=LogEvent(),
log_response=LogEvent(),
log_errors=LogEvent(),
)
重试
- 重试前
- 重试后
- 当重试耗尽时
GracefulRetry(
...,
log_before=LogEvent(),
log_after=LogEvent(),
log_exhausted=LogEvent(),
)
节流
- 当req/s限制达到时
- 当限制再次降低时
GracefulThrottle(
...,
log_limit_reached=LogEvent()
log_wait_over=LogEvent()
)
动态自定义
您可以通过传递一个lambda表达式进一步自定义它
LogEvent(
LogLevel.ERROR,
lambda r: "Request failed with {STATUS}" f" and it was {'redirected' if r.is_redirect else 'NOT redirected'}"
if r
else "",
)
请注意
- 并非所有日志事件都有响应,因此您需要保护自己免受其影响
- 占位符仍然有效(例如
{STATUS}
) - 您需要注意一些可能破坏格式化逻辑的属性(例如
r.headers
)
自定义异常
您可以定义自定义异常以获得更细粒度的控制权,有关更多信息,请参阅如何像专业人士一样结构Python中的异常。
您可以做的最简单的事情是
from gracy import Gracy, GracyConfig
from gracy.exceptions import GracyUserDefinedException
class MyCustomException(GracyUserDefinedException):
pass
class MyApi(Gracy[str]):
class Config:
SETTINGS = GracyConfig(
...,
parser={
HTTPStatus.BAD_REQUEST: MyCustomException
}
)
这将根据您的解析器中定义的条件抛出自定义异常。
您还可以通过自定义消息进一步改进它
class PokemonNotFound(GracyUserDefinedException):
BASE_MESSAGE = "Unable to find a pokemon with the name [{NAME}] at {URL} due to {STATUS} status"
def _format_message(self, request_context: GracyRequestContext, response: httpx.Response) -> str:
format_args = self._build_default_args()
name = request_context.endpoint_args.get("NAME", "Unknown")
return self.BASE_MESSAGE.format(NAME=name, **format_args)
报告
记录器
推荐用于生产环境。
Gracy使用logger.info
报告简短的摘要。
pokeapi = GracefulPokeAPI()
# do stuff with your API
pokeapi.report_status("logger")
# OUTPUT
❯ Gracy tracked that 'https://pokeapi.co/api/v2/pokemon/{NAME}' was hit 1 time(s) with a success rate of 100.00%, avg latency of 0.45s, and a rate of 1.0 reqs/s.
❯ Gracy tracked a total of 2 requests with a success rate of 100.00%, avg latency of 0.24s, and a rate of 1.0 reqs/s.
列表
使用print
生成所有属性的简短列表
pokeapi = GracefulPokeAPI()
# do stuff with your API
pokeapi.report_status("list")
# OUTPUT
____
/ ___|_ __ __ _ ___ _ _
| | _| '__/ _` |/ __| | | |
| |_| | | | (_| | (__| |_| |
\____|_| \__,_|\___|\__, |
|___/ Requests Summary Report
1. https://pokeapi.co/api/v2/pokemon/{NAME}
Total Reqs (#): 1
Success (%): 100.00%
Fail (%): 0.00%
Avg Latency (s): 0.39
Max Latency (s): 0.39
2xx Resps: 1
3xx Resps: 0
4xx Resps: 0
5xx Resps: 0
Avg Reqs/sec: 1.0 reqs/s
2. https://pokeapi.co/api/v2/generation/{ID}/
Total Reqs (#): 1
Success (%): 100.00%
Fail (%): 0.00%
Avg Latency (s): 0.04
Max Latency (s): 0.04
2xx Resps: 1
3xx Resps: 0
4xx Resps: 0
5xx Resps: 0
Avg Reqs/sec: 1.0 reqs/s
TOTAL
Total Reqs (#): 2
Success (%): 100.00%
Fail (%): 0.00%
Avg Latency (s): 0.21
Max Latency (s): 0.00
2xx Resps: 2
3xx Resps: 0
4xx Resps: 0
5xx Resps: 0
Avg Reqs/sec: 1.0 reqs/s
表格
它需要您安装Rich。
pokeapi = GracefulPokeAPI()
# do stuff with your API
pokeapi.report_status("rich")
以下是一个示例
Plotly
pokeapi = GracefulPokeAPI()
# do stuff with your API
plotly_fig = pokeapi.report_status("plotly")
plotly_fig.show()
以下是一个示例
重放请求
Gracy允许您重放先前交互中的请求和响应。
这是因为它允许您在没有延迟或消耗您的速率限制的情况下测试API。现在编写依赖于第三方API的单元测试是可行的。
它分为两步操作
步骤 | 描述 | 是否调用API? |
---|---|---|
4. 记录 | 存储所有请求/响应以供稍后重放 | 是 |
5. 重放 | 根据您的请求返回所有先前生成的响应作为“重放” | 否 |
记录
记录请求/响应的努力为零。您只需要将记录配置传递给您的Graceful API
from gracy import GracyReplay
from gracy.replays.storages.sqlite import SQLiteReplayStorage
record_mode = GracyReplay("record", SQLiteReplayStorage("pokeapi.sqlite3"))
pokeapi = GracefulPokeAPI(record_mode)
每个请求都将记录到定义的数据源。
重放
一旦您记录了所有请求,您就可以启用重放模式
from gracy import GracyReplay
from gracy.replays.storages.sqlite import SQLiteReplayStorage
replay_mode = GracyReplay("replay", SQLiteReplayStorage("pokeapi.sqlite3"))
pokeapi = GracefulPokeAPI(replay_mode)
每个请求都将路由到定义的数据源,从而实现更快的响应。
⚠️ 注意,解析器、重试、节流和类似的配置将按常规工作.
资源命名空间
您可以根据需要拥有多个命名空间来组织您的API端点。
为此,您只需从GracyNamespace
继承并在GracyAPI
中实例化它即可
from gracy import Gracy, GracyNamespace, GracyConfig
class PokemonNamespace(GracyNamespace[PokeApiEndpoint]):
async def get_one(self, name: str):
return await self.get(PokeApiEndpoint.GET_POKEMON, {"NAME": name})
class BerryNamespace(GracyNamespace[PokeApiEndpoint]):
async def get_one(self, name: str):
return await self.get(PokeApiEndpoint.GET_BERRY, {"NAME": name})
class GracefulPokeAPI(Gracy[PokeApiEndpoint]):
class Config:
BASE_URL = "https://pokeapi.co/api/v2/"
SETTINGS = GracyConfig(
retry=RETRY,
allowed_status_code={HTTPStatus.NOT_FOUND},
parser={HTTPStatus.NOT_FOUND: None},
)
# These will be automatically assigned on init
berry: BerryNamespace
pokemon: PokemonNamespace
使用方法如下
await pokeapi.pokemon.get_one("pikachu")
await pokeapi.berry.get_one("cheri")
请注意,所有配置都将传播到命名空间,但命名空间仍然可以有自己的配置,这在实例化时会引发合并。
分页
有些端点可能需要分页。为此,您可以使用GracyPaginator
。
对于只需传递offset
和limit
的简单情况,您可以使用GracyOffsetPaginator
from gracy import GracyOffsetPaginator
class BerryNamespace(GracyNamespace[PokeApiEndpoint]):
@parsed_response(ResourceList)
async def list(self, offset: int = 0, limit: int = 20):
params = dict(offset=offset, limit=limit)
return await self.get(PokeApiEndpoint.BERRY_LIST, params=params)
def paginate(self, limit: int = 20) -> GracyOffsetPaginator[ResourceList]:
return GracyOffsetPaginator[ResourceList](
gracy_func=self.list,
has_next=lambda r: bool(r["next"]) if r else True,
page_size=limit,
)
然后使用它
async def main():
api = PokeApi()
paginator = api.berry.paginate(2)
# Just grabs the next page
first = await paginator.next_page()
print(first)
# Resets current page to 0
paginator.set_page(0)
# Loop throught it all
async for page in paginator:
print(page)
高级使用
按方法自定义/覆盖配置
API可能根据端点返回不同的响应/条件/有效负载。
您可以通过使用@graceful
装饰器在每次方法的基础上覆盖任何GracyConfig
。
注意:如果您的函数使用yield
,请使用@graceful_generator
。
from gracy import Gracy, GracyConfig, GracefulRetry, graceful, graceful_generator
retry = GracefulRetry(...)
class GracefulPokeAPI(Gracy[PokeApiEndpoint]):
class Config:
BASE_URL = "https://pokeapi.co/api/v2/"
SETTINGS = GracyConfig(
retry=retry,
log_errors=LogEvent(
LogLevel.ERROR, "How can I become a pokemon master if {URL} keeps failing with {STATUS}"
),
)
@graceful(
retry=None, # 👈 Disables retry set in Config
log_errors=None, # 👈 Disables log_errors set in Config
allowed_status_code=HTTPStatus.NOT_FOUND,
parser={
"default": lambda r: r.json()["order"],
HTTPStatus.NOT_FOUND: None,
},
)
async def maybe_get_pokemon_order(self, name: str):
val: str | None = await self.get(PokeApiEndpoint.GET_POKEMON, {"NAME": name})
return val
@graceful( # 👈 Retry and log_errors are still set for this one
strict_status_code=HTTPStatus.OK,
parser={"default": lambda r: r.json()["order"]},
)
async def get_pokemon_order(self, name: str):
val: str = await self.get(PokeApiEndpoint.GET_POKEMON, {"NAME": name})
return val
@graceful_generator( # 👈 Retry and log_errors are still set for this one
parser={"default": lambda r: r.json()["order"]},
)
async def get_2_pokemons(self):
names = ["charmander", "pikachu"]
for name in names:
r = await self.get(PokeApiEndpoint.GET_POKEMON, {"NAME": name})
yield r
自定义HTTPx客户端
您可能想修改HTTPx客户端设置,通过以下方式实现
class YourAPIClient(Gracy[str]):
class Config:
...
def __init__(self, token: token) -> None:
self._token = token
super().__init__()
# 👇 Implement your logic here
def _create_client(self) -> httpx.AsyncClient:
client = super()._create_client()
client.headers = {"Authorization": f"token {self._token}"} # type: ignore
return client
覆盖默认请求超时
默认情况下,Gracy不会强制执行请求超时。
您可以在配置中设置自己的超时时间
class GracefulAPI(GracyApi[str]):
class Config:
BASE_URL = "https://example.com"
REQUEST_TIMEOUT = 10.2 # 👈 Here
创建自定义重放数据源
Gracy的设计理念是可扩展性。
您可以为存储/加载任何位置创建自己的存储(例如,SQL数据库),以下是一个示例
import httpx
from gracy import GracyReplayStorage
class MyCustomStorage(GracyReplayStorage):
def prepare(self) -> None: # (Optional) Executed upon API instance creation.
...
async def record(self, response: httpx.Response) -> None:
... # REQUIRED. Your logic to store the response object. Note the httpx.Response has request data.
async def _load(self, request: httpx.Request) -> httpx.Response:
... # REQUIRED. Your logic to load a response object based on the request.
# Usage
record_mode = GracyReplay("record", MyCustomStorage())
replay_mode = GracyReplay("replay", MyCustomStorage())
pokeapi = GracefulPokeAPI(record_mode)
请求前后钩子
您可以通过定义async def before
和async def after
方法来设置钩子。
⚠️ 注意:在这些方法中禁用了Gracy配置,这意味着重试/解析器/节流将不会在其中生效。
class GracefulPokeAPI(Gracy[PokeApiEndpoint]):
class Config:
BASE_URL = "https://pokeapi.co/api/v2/"
SETTINGS = GracyConfig(
retry=RETRY,
allowed_status_code={HTTPStatus.NOT_FOUND},
parser={HTTPStatus.NOT_FOUND: None},
)
def __init__(self, *args: t.Any, **kwargs: t.Any) -> None:
self.before_count = 0
self.after_status_counter = defaultdict[HTTPStatus, int](int)
self.after_aborts = 0
self.after_retries_counter = 0
super().__init__(*args, **kwargs)
async def before(self, context: GracyRequestContext):
self.before_count += 1
async def after(
self,
context: GracyRequestContext, # Current request context
response_or_exc: httpx.Response | Exception, # Either the request or an error
retry_state: GracefulRetryState | None, # Set when this is generated from a retry
):
if retry_state:
self.after_retries_counter += 1
if isinstance(response_or_exc, httpx.Response):
self.after_status_counter[HTTPStatus(response_or_exc.status_code)] += 1
else:
self.after_aborts += 1
async def get_pokemon(self, name: str):
return await self.get(PokeApiEndpoint.GET_POKEMON, {"NAME": name})
在上面的示例中,调用get_pokemon()
将按顺序触发before()
/after
钩子。
常见钩子
HttpHeaderRetryAfterBackOffHook
此钩子检查429(请求过多),然后读取retry-after
头。
如果设置了值,则Gracy将暂停所有客户端请求,直到时间过去。如果lock_per_endpoint
为True,则可以将此行为修改为按端点进行。
示例用法
from gracy.common_hooks import HttpHeaderRetryAfterBackOffHook
class GracefulAPI(GracyAPI[Endpoint]):
def __init__(self):
self._retry_after_hook = HttpHeaderRetryAfterBackOffHook(
self._reporter,
lock_per_endpoint=True,
log_event=LogEvent(
LogLevel.WARNING,
custom_message=(
"{ENDPOINT} produced {STATUS} and requested to wait {RETRY_AFTER}s "
"- waiting {RETRY_AFTER_ACTUAL_WAIT}s"
),
),
# Wait +10s to avoid this from happening again too soon
seconds_processor=lambda secs_requested: secs_requested + 10,
)
super().__init__()
async def before(self, context: GracyRequestContext):
await self._retry_after_hook.before(context)
async def after(
self,
context: GracyRequestContext,
response_or_exc: httpx.Response | Exception,
retry_state: GracefulRetryState | None,
):
retry_after_result = await self._retry_after_hook.after(context, response_or_exc)
RateLimitBackOffHook
此钩子检查429(请求过多)并锁定请求,时间为您定义的任意长度。
如果设置了值,则Gracy将暂停所有客户端请求,直到时间过去。如果lock_per_endpoint
为True,则可以将此行为修改为按端点进行。
from gracy.common_hooks import RateLimitBackOffHook
class GracefulAPI(GracyAPI[Endpoint]):
def __init__(self):
self._ratelimit_backoff_hook = RateLimitBackOffHook(
30,
self._reporter,
lock_per_endpoint=True,
log_event=LogEvent(
LogLevel.INFO,
custom_message="{UENDPOINT} got rate limited, waiting for {WAIT_TIME}s",
),
)
super().__init__()
async def before(self, context: GracyRequestContext):
await self._ratelimit_backoff_hook.before(context)
async def after(
self,
context: GracyRequestContext,
response_or_exc: httpx.Response | Exception,
retry_state: GracefulRetryState | None,
):
backoff_result = await self._ratelimit_backoff_hook.after(context, response_or_exc)
from gracy.common_hooks import HttpHeaderRetryAfterBackOffHook, RateLimitBackOffHook
📚 额外资源
过去几年中我学到的良好做法指导了Gracy的哲学,您可能会从中受益
变更日志
查看变更日志。
许可证
MIT
致谢
感谢我最后三个工作过的初创公司,它们迫使我一次又一次地做同样的事情和解决同样的问题。我对此感到厌倦,并构建了这个库。
最重要的是:感谢上帝,他允许我(一个随意的🇧🇷人)为许多不同的🇺🇸初创公司工作。这是讽刺的,因为由于上帝的恩典,我能够构建Gracy。🙌
项目详情
下载文件
下载您平台对应的文件。如果您不确定选择哪个,请了解有关安装包的更多信息。
源代码分发
构建分发
gracy-1.33.0.tar.gz的哈希值
算法 | 哈希摘要 | |
---|---|---|
SHA256 | fb76070dc5f70b3bb50c71af091dedd10c6b14e8d56559d9e8ba7cff5ae58768 |
|
MD5 | b996fa61e681911c2e1a83f956881872 |
|
BLAKE2b-256 | d64221f3caf7448135a134720f65366430bf6e13e6f03c15977aca5693575d85 |
gracy-1.33.0-py3-none-any.whl的哈希值
算法 | 哈希摘要 | |
---|---|---|
SHA256 | 41f7ea136ce2d3507381ca287ed4f97e486044258e16ae8a4b4db35c2175047e |
|
MD5 | 484367be30fc32e97a2b9e99ac4ce42a |
|
BLAKE2b-256 | 62351cd65888de6df73e7d657b3df6fe62373e0ba235bb337fb1ea55e95dd220 |