HTTSleep

Common errors

• AttributeError: 'NoneType' object has no attribute 'json'

  cause The `http_get_func` provided to `HTTSleep` returned `None` or an object that does not have a `.json()` method (or other expected `requests.Response` attributes), typically when the underlying HTTP request failed or was improperly mocked.
  fix Verify that your `http_get_func` always returns a `requests.Response`-like object, even in error scenarios (e.g., by catching exceptions and returning a `requests.Response` with an error status code and empty content), or ensure your mock is complete.

• httsleep.exceptions.HTTSleepTimeout: Polling timed out after X seconds.

  cause The `check_func` never returned `True` within the total `timeout` period specified. This can be due to the target service taking too long, the success condition never being met, or the `check_func` having a bug.
  fix Increase the `timeout` value, adjust `interval`, `max_interval`, and `factor` to allow for longer and more frequent retries, or thoroughly debug your `check_func` and the target service to ensure the success condition is eventually reachable and correctly identified.

• ModuleNotFoundError: No module named 'httsleep'

  cause The `httsleep` library is not installed in the current Python environment, or the environment where the code is being run is different from where `pip install httsleep` was executed.
  fix Ensure `httsleep` is installed by running `pip install httsleep`. If you are using a virtual environment, activate it before running the installation and your script.

Warnings

• gotcha HTTSleep has not been actively maintained or updated since January 2021. This means it may not be compatible with newer Python versions (beyond 3.8/3.9, which were current at the time of its last update), or its dependencies might become outdated, potentially leading to security vulnerabilities or runtime errors.
  Fix: Thoroughly test `httsleep` in your target Python environment. Consider forking the project or choosing a more actively maintained alternative if long-term support and up-to-date dependencies are critical.
• gotcha The `http_get_func` passed to `HTTSleep` must return an object that mimics `requests.Response`, specifically having a `.json()` method and a `status_code` attribute, if these are used by your `check_func` or the default status checkers. Returning a non-compatible object (e.g., `None` or a plain dictionary) will lead to `AttributeError` or `TypeError`.
  Fix: Ensure your custom `http_get_func` uses `requests.get` (or similar `requests` methods) or constructs a `requests.Response`-compatible object. If you're mocking, ensure the mock object has the expected attributes and methods like `.json()` and `status_code`.
• gotcha HTTSleep uses `python-json-logger` for its internal logging. If you don't configure the logger, it may log to the console with JSON formatting by default, which might be unexpected in non-JSON logging setups. You can explicitly pass `logger=None` or configure a custom logger.
  Fix: To control logging, either pass `logger=None` to the `HTTSleep` constructor to disable it, or provide a pre-configured `logging.Logger` instance: `logger=logging.getLogger('my_logger')`.
• gotcha HTTSleep provides default status checking logic based on HTTP status codes (e.g., 200 for success, 2xx for continuation). If you define a `check_func`, it completely overrides these defaults. Ensure your `check_func` handles all conditions, including what constitutes a success and what should trigger a retry or failure.
  Fix: If you need both custom logic and default HTTP status checks, incorporate the default logic into your `check_func`, or chain it. Alternatively, consider using `status_checker` argument if applicable, but be aware of how they interact.

Install

• pip install httsleep Install httsleep

Imports

• HTTSleep

  from httsleep import HTTSleep

Quickstart

This quickstart demonstrates how to use `HTTSleep` to poll an endpoint until a specific condition is met. The `simulate_api_call` function acts as a mock HTTP service, returning 'pending' status a few times before eventually returning 'completed'. The `check_func` defines the condition for successful polling. The poller will retry with exponential backoff until the condition is met or the `timeout` is reached.

import time
from httsleep import HTTSleep
import requests # To create a mock response for the example
import json

# Simulate a remote service that eventually returns 'completed'
# In a real scenario, this would be your actual API endpoint.
mock_response_count = 0
def simulate_api_call(url=None, **kwargs):
    nonlocal mock_response_count
    mock_response_count += 1

    mock_resp = requests.Response()
    if mock_response_count < 3:
        mock_resp.status_code = 202 # Accepted (still processing)
        mock_resp._content = b'{"status": "pending", "detail": "processing..."}'
        mock_resp.request = requests.Request('GET', url or 'http://mock.api/status').prepare()
    else:
        mock_resp.status_code = 200 # OK
        mock_resp._content = b'{"status": "completed", "result": "success!"}'
        mock_resp.request = requests.Request('GET', url or 'http://mock.api/status').prepare()
    
    # Monkey-patch json method for simplicity
    def json_method():
        return json.loads(mock_resp._content.decode('utf-8'))
    mock_resp.json = json_method
    return mock_resp

# Create an HTTSleep instance
poller = HTTSleep(
    http_get_func=simulate_api_call, # Pass the mock function that makes HTTP calls
    check_func=lambda resp: resp.json().get("status") == "completed",
    timeout=20, # Total timeout for polling in seconds
    interval=1, # Initial interval in seconds
    max_interval=5, # Max interval between retries
    factor=2, # Exponential backoff factor
    logger=None # Suppress default logging for clean quickstart output
)

print("Starting polling for a 'completed' status...")
try:
    final_response = poller.poll()
    print(f"Polling successful! Final status: {final_response.json().get('status')}")
    print(f"Total attempts: {mock_response_count}")
except Exception as e:
    print(f"Polling failed after timeout: {e}")

# Reset count for potential re-runs in an interactive environment
mock_response_count = 0