Skip to main content

Gracefully manage your API interactions

Project description

Gracefully manage your API interactions

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

Gracy helps you handle failures, logging, retries, throttling, and tracking for all your HTTP interactions. Gracy uses httpx under the hood.

"Let Gracy do the boring stuff while you focus on your application"


Summary

🧑‍💻 Get started

Installation

pip install gracy

OR

poetry add gracy

Usage

Example will use the PokeAPI.

Simple example

# 0. Import
import asyncio
from typing import Awaitable
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[PokeApiEndpoint]):
    class Config:  # type: ignore
        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) -> 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()


asyncio.run(main())

More examples

Settings

Strict/Allowed status code

By default Gracy considers any successful status code (200-299) as successful.

Strict

You can modify this behavior by defining a strict status code or increase the range of allowed status codes:

from http import HTTPStatus

GracyConfig(
  strict_status_code=HTTPStatus.CREATED
)

or a list of values:

from http import HTTPStatus

GracyConfig(
  strict_status_code={HTTPStatus.OK, HTTPStatus.CREATED}
)

Using strict_status_code means that any other code no specified will raise an error regardless of being successful or not.

Allowed

You can also keep the behavior, but extend the range of allowed codes.

from http import HTTPStatus

GracyConfig(
  allowed_status_code=HTTPStatus.NOT_FOUND
)

or a list of values

from http import HTTPStatus

GracyConfig(
  allowed_status_code={HTTPStatus.NOT_FOUND, HTTPStatus.FORBIDDEN}
)

Using allowed_status_code means that all successful codes plus your defined codes will be considered successful.

This is quite useful for parsing as you'll see soon.

⚠️ Note that strict_status_code takes precedence over allowed_status_code, probably you don't want to combine those. Prefer one or the other.

Parsing

Parsing allows you to handle the request based on the status code returned.

The basic example is parsing json:

GracyConfig(
  parser={
    "default": lambda r: r.json()
  }
)

In this example all successful requests will automatically return the json() result.

You can also narrow it down to handle specific status codes.

class Config:
  GracyConfig(
    ...,
    allowed_status_code: HTTPStatusCode.NOT_FOUND,
    parser={
      "default": lambda r: r.json()
      HTTPStatusCode.NOT_FOUND: None
    }
  )

async def get_pokemon(self, name: str) -> Awaitable[dict| None]:
  # 👇 Returns either dict or None
  return await self.get(PokeApiEndpoint.GET_POKEMON, {"NAME": name})

Or even customize exceptions to improve your code readability:

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})

Retry

Who doesn't hate flaky APIs? 🙋

Yet there're many of them.

Using tenacity, backoff, retry, aiohttp_retry, and any other retry libs is NOT easy enough. 🙅

You still would need to code the implementation for each request which is annoying.

Here's how Gracy allows you to gracefully code your retry logic:

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",
    )
  )
Parameter Description Example
delay How many seconds to wait between retries 2 would wait 2 seconds, 1.5 would wait 1.5 seconds, and so on
max_attempts How many times should Gracy retry the request? 10 means 1 regular request with additional 10 retries in case they keep failing. 1 should be the minimum
delay_modifier Allows you to specify increasing delay times by multiplying this value to delay Setting 1 means no delay change. Setting 2 means delay will be doubled every retry
retry_on Should we retry for which status codes? None means for any non successful status code HTTPStatus.BAD_REQUEST, or {HTTPStatus.BAD_REQUEST, HTTPStatus.FORBIDDEN}
log_before Specify log level. None means don't log More on logging later
log_after Specify log level. None means don't log More on logging later
log_exhausted Specify log level. None means don't log More on logging later
behavior Allows you to define how to deal if the retry fails. pass will accept any retry failure pass or break (default)

Throttling

Rate limiting issues? No more.

Gracy helps you proactively deal with it before any API throws 429 in your face.

Creating rules

You can define rules per endpoint using regex:

TWO_REQS_FOR_ANY_ENDPOINT_RULE = ThrottleRule(
  url_pattern=r".*",
  requests_per_second_limit=2
)

TEN_REQS_FOR_ANY_POKEMON_ENDPOINT_RULE = ThrottleRule(
  url_pattern=r".*\/pokemon\/.*",
  requests_per_second_limit=10
)

Setting throttling

You can set up logging and assign rules as:

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),
    ),
  )

Logging

You can define and customize logs for events by using LogEvent and 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")

Note that placeholders are formatted and replaced later on by Gracy based on the event type, like:

Placeholders per event

Placeholder Description Example Supported Events
{URL} Endpoint being targetted. Sometimes formatted, sometimes containg placeholders. /pokemon/{NAME}, /pokemon/{PIKACHU}
  • [x] Before request
  • [x] After response
  • [x] Response error
  • [x] Before retry
  • [x] After retry
  • [x] Retry exhausted
  • [x] Throttle limit reached
  • [x] Throttle limit available
{METHOD} HTTP Request being used GET, POST
  • [x] Before request
  • [x] After response
  • [x] Response error
  • [ ] Before retry
  • [ ] After retry
  • [ ] Retry exhausted
  • [ ] Throttle limit reached
  • [ ] Throttle limit available
{STATUS} Status code returned by the response 200, 404, 501
  • [ ] Before request
  • [x] After response
  • [x] Response error
  • [ ] Before retry
  • [ ] After retry
  • [ ] Retry exhausted
  • [ ] Throttle limit reached
  • [ ] Throttle limit available
{ELAPSED} Amount of seconds taken for the request to complete Numeric
  • [ ] Before request
  • [x] After response
  • [x] Response error
  • [ ] Before retry
  • [ ] After retry
  • [ ] Retry exhausted
  • [ ] Throttle limit reached
  • [ ] Throttle limit available
{RETRY_DELAY} How long Gracy will wait before repeating the request Numeric Any Retry event
{CUR_ATTEMPT} Current attempt count for the current request Numeric Any Retry event
{MAX_ATTEMPT} Max attempt defined for the current request Numeric Any Retry event
{THROTTLE_LIMIT} How many reqs/s is defined for the current request Numeric Any Throttle event
{THROTTLE_TIME} How long Gracy will wait before calling the request Numeric Any Throttle event

and you can set up the log events as follows:

Requests

  1. Before request
  2. After response
  3. Response has non successful errors
GracyConfig(
  log_request=LogEvent(),
  log_response=LogEvent(),
  log_errors=LogEvent(),
)

Retry

  1. Before retry
  2. After retry
  3. When retry exhausted
GracefulRetry(
  ...,
  log_before=LogEvent(),
  log_after=LogEvent(),
  log_exhausted=LogEvent(),
)

Throttling

  1. When reqs/s limit is reached
  2. When limit decreases again
GracefulThrottle(
  ...,
  log_limit_reached=LogEvent()
  log_wait_over=LogEvent()
)

Custom Exceptions

You can define custom exceptions for more fine grained control over your exception messages/types.

The simplest you can do is:

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
      }
    )

This will raise your custom exception under the conditions defined in your parser.

class PokemonNotFound(GracyUserDefinedException):
    BASE_MESSAGE = "Unable to find a pokemon with the name [{NAME}] at {URL} due to {STATUS} status"

    def _format_message(self, base_endpoint: str, endpoint_args: dict[str, str], response: httpx.Response) -> str:
        format_args = self._build_default_args()
        name = endpoint_args.get("NAME", "Unknown")
        return self.BASE_MESSAGE.format(NAME=name, **format_args)

Reports

Generate Rich reports and find out:

  • Your API success/fail rate
  • Average latency
  • Slowest endpoints
  • Times an endpoint got hit
  • Status group returned
class GracefulPokeAPI(Gracy[PokeApiEndpoint]):
  ...

pokeapi = GracefulPokeAPI()
# do stuff with your API
pokeapi.report_status()

Here's an example of how it looks:

Report

Advanced Usage

Customizing/Overriding configs per method

APIs may return different responses/conditions/payloads based on the endpoint.

You can override any GracyConfig on a per method basis by using the graceful decorator.

from gracy import Gracy, GracyConfig, GracefulRetry, graceful

retry = GracefulRetry(...)

class GracefulPokeAPI(Gracy[PokeApiEndpoint]):
    class Config:  # type: ignore
        BASE_URL = "https://pokeapi.co/api/v2/"
        SETTINGS = GracyConfig(
            retry=retry,
            log_errors=LogEvent(
                LogLevel.ERROR, "How can I become a master pokemon 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

Customizing HTTPx client

You might want to modify the HTTPx client settings, do so by:

class YourAPIClient(Gracy[str]):
    class Config:  # type: ignore
        ...

    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

📚 Extra Resources

Some good practices I learned over the past years guided Gracy's philosophy, you might benefit by reading:

Change log

See CHANGELOG.

License

MIT

Credits

Thanks to the last three startups I worked which forced me to do the same things and resolve the same problems over and over again. I got sick of it and built this lib.

Most importantly: Thanks to God, who allowed me (a random 🇧🇷 guy) to work for many different 🇺🇸 startups. This is ironic since due to God's grace, I was able to build Gracy. 🙌

Also, thanks to the httpx and rich projects for the beautiful and simple APIs that powers Gracy.

Project details


Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Source Distribution

gracy-1.1.0.tar.gz (23.1 kB view hashes)

Uploaded Source

Built Distribution

gracy-1.1.0-py3-none-any.whl (18.9 kB view hashes)

Uploaded Python 3

Supported by

AWS AWS Cloud computing and Security Sponsor Datadog Datadog Monitoring Fastly Fastly CDN Google Google Download Analytics Microsoft Microsoft PSF Sponsor Pingdom Pingdom Monitoring Sentry Sentry Error logging StatusPage StatusPage Status page