Invoker Guard

A simple, transport-agnostic wrapper for executing callables with robust, configurable retry logic, powered by tenacity.

It allows you to write your retry and validation logic once and apply it to different HTTP libraries like requests, httpx, and niquests.

Installation

Install the core library:

pip install invoker-guard

Then, install the support for the HTTP library you want to use:

# For requests support
pip install invoker-guard[requests]

# For httpx support
pip install invoker-guard[httpx]

# For everything
pip install invoker-guard[requests,httpx,niquests]

How to Use

The library's main function is invoke. You provide it with:

1. A callable_action (a lambda that calls your HTTP library).
2. A RetryConfig object defining your rules.
3. A Transport object that matches your library.

Example with requests

import requests
from invoker_guard import invoke, RetryConfig, RetryError, REQUESTS_TRANSPORT

# Define rules
config = RetryConfig(expected_status_codes=[200])

# Define action
url = "https://api.example.com/data"
action = lambda: requests.get(url, timeout=10)

# Invoke it
try:
    response = invoke(action, config, transport=REQUESTS_TRANSPORT)
    print("Success:", response.json())
except RetryError as e:
    print("Failed after all retries:", e)

Example with httpx

The logic is identical—just swap the library and the transport object.

import httpx
from invoker_guard import invoke, RetryConfig, RetryError, HTTPX_TRANSPORT

# Define rules (can be the same config object)
config = RetryConfig(expected_status_codes=[200], required_json_keys=["id"])

# Define action for httpx
url = "https://api.example.com/data"
action = lambda: httpx.get(url, timeout=10)

# Invoke it
try:
    response = invoke(action, config, transport=HTTPX_TRANSPORT)
    print("Success:", response.json())
except RetryError as e:
    print("Failed after all retries:", e)