Skip to content

fix(retry): never replay a non-idempotent write; bound server retry signals (#104) #115

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
karlwaldman merged 1 commit into from
Sep 13, 2026
Merged

fix(retry): never replay a non-idempotent write; bound server retry signals (#104) #115

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
57 changes: 44 additions & 13 deletions oilpriceapi/async_client.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -44,7 +44,7 @@
)
from .models import HistoricalPrice, HistoricalResponse, MarketBrief, Price
from .resource_validators import format_date
from .retry import RetryStrategy
from .retry import RetryStrategy, mark_ambiguous_write, validated_max_retries


class AsyncOilPriceAPI:
Expand All @@ -56,7 +56,13 @@ class AsyncOilPriceAPI:
api_key: API key for authentication
base_url: Base URL for API
timeout: Request timeout in seconds
max_retries: Maximum request attempts
max_retries: Total request ATTEMPTS, not retries after the first.
Must be at least 1; anything less raises ConfigurationError.
retry_on: Status codes to retry on. An explicit empty list is honoured.

Retry safety: POST and PATCH are sent exactly once -- a timeout, transport
error or 5xx is ambiguous, so replaying could duplicate the write. Pass
``idempotent=True`` to ``request()`` to opt back in.

Example:
>>> async with AsyncOilPriceAPI() as client:
Expand Down Expand Up @@ -93,8 +99,13 @@ def __init__(
# Configuration
self.base_url = (base_url or self.DEFAULT_BASE_URL).rstrip("/")
self.timeout = timeout or self.DEFAULT_TIMEOUT
self.max_retries = max_retries or self.DEFAULT_MAX_RETRIES
self.retry_on = retry_on or self.DEFAULT_RETRY_CODES
# Explicit None checks, not `or` (#104) -- see OilPriceAPI.__init__.
self.max_retries = (
self.DEFAULT_MAX_RETRIES if max_retries is None else validated_max_retries(max_retries)
)
self.retry_on = (
list(self.DEFAULT_RETRY_CODES) if retry_on is None else list(retry_on)
)
self.max_connections = max_connections
self.max_keepalive_connections = max_keepalive_connections
self.app_url = app_url
Expand Down Expand Up @@ -196,9 +207,16 @@ async def request(
path: str,
params: Optional[Dict[str, Any]] = None,
json_data: Optional[Dict[str, Any]] = None,
idempotent: Optional[bool] = None,
**kwargs,
) -> Union[Dict[str, Any], List[Any]]:
"""Make async HTTP request to API."""
"""Make async HTTP request to API.

Args:
idempotent: Assert that repeating this request is safe. Without it,
a non-idempotent method (POST, PATCH) is sent exactly once and
never replayed after an ambiguous outcome (#104).
"""
await self._ensure_client()
assert self._client is not None # set by _ensure_client

Expand Down Expand Up @@ -239,19 +257,24 @@ async def request(
)

# Auto-retry with Retry-After if we have attempts left
if self._retry_strategy.should_retry(attempt, 429, response.headers):
try:
wait_time = min(float(retry_after), 60.0)
except (TypeError, ValueError):
wait_time = self._retry_strategy.calculate_wait_time(attempt)
if self._retry_strategy.should_retry(
attempt, 429, response.headers, method=method, idempotent=idempotent
):
wait_time = self._retry_strategy.bounded_wait(
retry_after, self._retry_strategy.calculate_wait_time(attempt)
)
logger.info(
f"Rate limited. Retrying in {wait_time}s (attempt {attempt + 1}/{self.max_retries})"
)
await asyncio.sleep(wait_time)
continue
elif response.status_code >= 500:
if self._retry_strategy.should_retry(
attempt, response.status_code, response.headers
attempt,
response.status_code,
response.headers,
method=method,
idempotent=idempotent,
):
wait_time = self._retry_strategy.calculate_wait_time(attempt)
self._retry_strategy.log_retry(
Expand All @@ -273,20 +296,26 @@ async def request(
api_key=self.api_key,
timeout=self.timeout,
)
if self._retry_strategy.should_retry_on_exception(attempt):
if self._retry_strategy.should_retry_on_exception(
attempt, method=method, idempotent=idempotent
):
wait_time = self._retry_strategy.calculate_wait_time(attempt)
self._retry_strategy.log_retry(
attempt, "Request timeout", wait_time, is_async=True
)
await asyncio.sleep(wait_time)
continue
if not self._retry_strategy.is_replay_safe(method, idempotent):
raise mark_ambiguous_write(last_exception, method)
raise last_exception
except httpx.RequestError as error:
last_exception = error_from_exception(
error,
api_key=self.api_key,
)
if self._retry_strategy.should_retry_on_exception(attempt):
if self._retry_strategy.should_retry_on_exception(
attempt, method=method, idempotent=idempotent
):
wait_time = self._retry_strategy.calculate_wait_time(attempt)
self._retry_strategy.log_retry(
attempt,
Expand All @@ -296,6 +325,8 @@ async def request(
)
await asyncio.sleep(wait_time)
continue
if not self._retry_strategy.is_replay_safe(method, idempotent):
raise mark_ambiguous_write(last_exception, method)
raise last_exception

if last_exception:
Expand Down
100 changes: 80 additions & 20 deletions oilpriceapi/client.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -45,7 +45,7 @@
from .resources.subscriptions import SubscriptionsResource
from .resources.webhooks import WebhooksResource
from .resources.well_production import WellProductionResource
from .retry import RetryStrategy
from .retry import RetryStrategy, mark_ambiguous_write, validated_max_retries


class OilPriceAPI:
Expand All @@ -63,8 +63,20 @@ class OilPriceAPI:
api_key: API key for authentication. If not provided, uses OILPRICEAPI_KEY env var.
base_url: Base URL for API. Defaults to production.
timeout: Request timeout in seconds. Defaults to 30.
max_retries: Maximum request attempts for failed requests. Defaults to 3.
max_retries: Total request ATTEMPTS, not retries after the first.
Defaults to 3. Must be at least 1; pass 1 for a single attempt with
no retries. Anything less, or a non-int, raises ConfigurationError
rather than being silently replaced by the default.
retry_on: Status codes to retry on. Defaults to [429, 500, 502, 503, 504].
An explicit empty list is honoured and disables status-code retries.

Retry safety: a non-idempotent method (POST, PATCH) is sent exactly ONCE.
A timeout, a transport error or a 5xx is an ambiguous outcome — the server
may have committed the write before the response was lost — so replaying it
could create a duplicate. Idempotent methods (GET, HEAD, OPTIONS, TRACE,
PUT, DELETE) still retry as before, and a 429 still retries any method
because it is an outright refusal. Pass ``idempotent=True`` to ``request()``
to opt a specific write back into retrying.

Example:
>>> # Recommended: Use context manager for automatic cleanup
Expand Down Expand Up @@ -108,8 +120,15 @@ def __init__(
# Configuration
self.base_url = (base_url or self.DEFAULT_BASE_URL).rstrip("/")
self.timeout = timeout or self.DEFAULT_TIMEOUT
self.max_retries = max_retries or self.DEFAULT_MAX_RETRIES
self.retry_on = retry_on or self.DEFAULT_RETRY_CODES
# Explicit None checks, not `or`: an explicit max_retries=0 used to
# become 3 and an explicit retry_on=[] used to become the default status
# list, silently discarding what the caller asked for (#104).
self.max_retries = (
self.DEFAULT_MAX_RETRIES if max_retries is None else validated_max_retries(max_retries)
)
self.retry_on = (
list(self.DEFAULT_RETRY_CODES) if retry_on is None else list(retry_on)
)

# Initialize retry strategy
self._retry_strategy = RetryStrategy(max_retries=self.max_retries, retry_on=self.retry_on)
Expand Down Expand Up @@ -205,6 +224,7 @@ def request(
params: Optional[Dict[str, Any]] = None,
json_data: Optional[Dict[str, Any]] = None,
timeout: Optional[float] = None,
idempotent: Optional[bool] = None,
**kwargs,
) -> Dict[str, Any]:
"""Make HTTP request to API.
Expand All @@ -218,6 +238,9 @@ def request(
params: Query parameters
json_data: JSON body data
timeout: Request timeout in seconds. If None, uses client's default timeout.
idempotent: Assert that repeating this request is safe. Without it,
POST and PATCH are sent exactly once and never replayed after an
ambiguous outcome (#104).
**kwargs: Additional httpx request arguments

Returns:
Expand Down Expand Up @@ -274,18 +297,28 @@ def request(
)

# Auto-retry with Retry-After if we have attempts left
if self._retry_strategy.should_retry(attempt, 429, response.headers):
try:
wait_time = min(float(retry_after), 60.0)
except (TypeError, ValueError):
wait_time = self._retry_strategy.calculate_wait_time(attempt)
if self._retry_strategy.should_retry(
attempt, 429, response.headers, method=method, idempotent=idempotent
):
# Bounded in BOTH directions: a server Retry-After of
# 31612 would park the process for 8.8 hours, and a
# negative one makes time.sleep() raise (#104).
wait_time = self._retry_strategy.bounded_wait(
retry_after, self._retry_strategy.calculate_wait_time(attempt)
)
logger.info(
f"Rate limited. Retrying in {wait_time}s (attempt {attempt + 1}/{self.max_retries})"
)
time.sleep(wait_time)
continue
elif response.status_code >= 500:
if self._retry_strategy.should_retry(attempt, response.status_code, response.headers):
if self._retry_strategy.should_retry(
attempt,
response.status_code,
response.headers,
method=method,
idempotent=idempotent,
):
wait_time = self._retry_strategy.calculate_wait_time(attempt)
self._retry_strategy.log_retry(
attempt,
Expand All @@ -306,21 +339,27 @@ def request(
api_key=self.api_key,
timeout=effective_timeout,
)
if self._retry_strategy.should_retry_on_exception(attempt):
if self._retry_strategy.should_retry_on_exception(
attempt, method=method, idempotent=idempotent
):
wait_time = self._retry_strategy.calculate_wait_time(attempt)
self._retry_strategy.log_retry(
attempt, "Request timeout", wait_time, is_async=False
)
time.sleep(wait_time)
continue
if not self._retry_strategy.is_replay_safe(method, idempotent):
raise mark_ambiguous_write(last_exception, method)
logger.error(f"Request timed out after {self.max_retries} attempts")
raise last_exception
except httpx.RequestError as error:
last_exception = error_from_exception(
error,
api_key=self.api_key,
)
if self._retry_strategy.should_retry_on_exception(attempt):
if self._retry_strategy.should_retry_on_exception(
attempt, method=method, idempotent=idempotent
):
wait_time = self._retry_strategy.calculate_wait_time(attempt)
self._retry_strategy.log_retry(
attempt,
Expand All @@ -330,6 +369,8 @@ def request(
)
time.sleep(wait_time)
continue
if not self._retry_strategy.is_replay_safe(method, idempotent):
raise mark_ambiguous_write(last_exception, method)
logger.error(
f"Request failed after {self.max_retries} attempts: "
f"{error.__class__.__name__}"
Expand All @@ -354,6 +395,7 @@ def request_with_headers(
params: Optional[Dict[str, Any]] = None,
json_data: Optional[Dict[str, Any]] = None,
timeout: Optional[float] = None,
idempotent: Optional[bool] = None,
**kwargs,
) -> Tuple[Dict[str, Any], httpx.Headers]:
"""Make HTTP request and return (json_body, headers) tuple.
Expand Down Expand Up @@ -388,18 +430,28 @@ def request_with_headers(
if response.status_code == 429:
retry_after = response.headers.get("Retry-After")

if self._retry_strategy.should_retry(attempt, 429, response.headers):
try:
wait_time = min(float(retry_after), 60.0)
except (TypeError, ValueError):
wait_time = self._retry_strategy.calculate_wait_time(attempt)
if self._retry_strategy.should_retry(
attempt, 429, response.headers, method=method, idempotent=idempotent
):
# Bounded in BOTH directions: a server Retry-After of
# 31612 would park the process for 8.8 hours, and a
# negative one makes time.sleep() raise (#104).
wait_time = self._retry_strategy.bounded_wait(
retry_after, self._retry_strategy.calculate_wait_time(attempt)
)
logger.info(
f"Rate limited. Retrying in {wait_time}s (attempt {attempt + 1}/{self.max_retries})"
)
time.sleep(wait_time)
continue
elif response.status_code >= 500:
if self._retry_strategy.should_retry(attempt, response.status_code, response.headers):
if self._retry_strategy.should_retry(
attempt,
response.status_code,
response.headers,
method=method,
idempotent=idempotent,
):
wait_time = self._retry_strategy.calculate_wait_time(attempt)
self._retry_strategy.log_retry(
attempt,
Expand All @@ -420,20 +472,26 @@ def request_with_headers(
api_key=self.api_key,
timeout=effective_timeout,
)
if self._retry_strategy.should_retry_on_exception(attempt):
if self._retry_strategy.should_retry_on_exception(
attempt, method=method, idempotent=idempotent
):
wait_time = self._retry_strategy.calculate_wait_time(attempt)
self._retry_strategy.log_retry(
attempt, "Request timeout", wait_time, is_async=False
)
time.sleep(wait_time)
continue
if not self._retry_strategy.is_replay_safe(method, idempotent):
raise mark_ambiguous_write(last_exception, method)
raise last_exception
except httpx.RequestError as error:
last_exception = error_from_exception(
error,
api_key=self.api_key,
)
if self._retry_strategy.should_retry_on_exception(attempt):
if self._retry_strategy.should_retry_on_exception(
attempt, method=method, idempotent=idempotent
):
wait_time = self._retry_strategy.calculate_wait_time(attempt)
self._retry_strategy.log_retry(
attempt,
Expand All @@ -443,6 +501,8 @@ def request_with_headers(
)
time.sleep(wait_time)
continue
if not self._retry_strategy.is_replay_safe(method, idempotent):
raise mark_ambiguous_write(last_exception, method)
raise last_exception

if last_exception:
Expand Down
Loading