跳转到主要内容

优雅地管理您的API交互

项目描述

Python最优雅的API客户端框架

Actions Status PyPI Downloads GitHub Code style: black try/except style: tryceratops Types: pyright Follow guilatrova Sponsor guilatrova

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 将接受任何重试失败 passbreak(默认值)
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):
      ...

日志记录

您可以使用LogEventLogLevel定义和自定义日志事件。

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} 节流规则定义的时间范围 second90 seconds 任何节流事件

您可以根据以下方式设置日志事件

请求

  1. 请求前
  2. 响应后
  3. 响应有非成功的错误
GracyConfig(
  log_request=LogEvent(),
  log_response=LogEvent(),
  log_errors=LogEvent(),
)

重试

  1. 重试前
  2. 重试后
  3. 当重试耗尽时
GracefulRetry(
  ...,
  log_before=LogEvent(),
  log_after=LogEvent(),
  log_exhausted=LogEvent(),
)

节流

  1. 当req/s限制达到时
  2. 当限制再次降低时
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")

以下是一个示例

Report

Plotly

它需要您安装plotly 📊pandas 🐼

pokeapi = GracefulPokeAPI()
# do stuff with your API
plotly_fig = pokeapi.report_status("plotly")
plotly_fig.show()

以下是一个示例

Report

重放请求

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

对于只需传递offsetlimit的简单情况,您可以使用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 beforeasync 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。🙌

此外,感谢httpxrich项目为Gracy提供了美丽而简单的API。

项目详情


下载文件

下载您平台对应的文件。如果您不确定选择哪个,请了解有关安装包的更多信息。

源代码分发

gracy-1.33.0.tar.gz (48.4 kB 查看哈希值)

上传时间 源代码

构建分发

gracy-1.33.0-py3-none-any.whl (46.1 kB 查看哈希值)

上传时间 Python 3

由以下机构支持

AWS AWS 云计算和安全赞助商 Datadog Datadog 监控 Fastly Fastly CDN Google Google 下载分析 Microsoft Microsoft PSF 赞助商 Pingdom Pingdom 监控 Sentry Sentry 错误记录 StatusPage StatusPage 状态页面