Last active
January 26, 2026 04:38
-
-
Save samukasmk/c426b033a0b6943f0cf0a2bd83139299 to your computer and use it in GitHub Desktop.
Python requests - HTTPAdapter - Retry when received (HTTP 429 Retry-After) by static time
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| import logging | |
| import requests | |
| from requests.adapters import HTTPAdapter | |
| from urllib3.util.retry import Retry | |
| log = logging.getLogger("http.retry") | |
| class FixedDelayLoggingRetry(Retry): | |
| def get_backoff_time(self) -> float: | |
| return 1.0 # always 1s between retries | |
| def increment(self, method=None, url=None, response=None, error=None, _pool=None, _stacktrace=None): | |
| new_retry = super().increment( | |
| method=method, url=url, response=response, error=error, _pool=_pool, _stacktrace=_stacktrace | |
| ) | |
| # history: list of failures that triggered retries | |
| retry_count = len(new_retry.history) # how many retries already happened | |
| next_attempt = retry_count + 1 # the next attempt that will run (1st, 2nd, 3rd...) | |
| status = getattr(response, "status", None) if response is not None else None | |
| reason = f"status={status}" if status is not None else f"error={type(error).__name__}" if error else "unknown" | |
| log.warning( | |
| "Retry triggered: method=%s url=%s reason=%s -> next attempt #%d in %.1fs (remaining total=%s)", | |
| method, url, reason, next_attempt, self.get_backoff_time(), getattr(new_retry, "total", None) | |
| ) | |
| return new_retry | |
| def build_session(): | |
| retry = FixedDelayLoggingRetry( | |
| total=9, | |
| status_forcelist=(408, 429, 502, 503, 504), | |
| allowed_methods=("GET", "POST", "PUT", "PATCH", "DELETE"), | |
| respect_retry_after_header=False, # if True, 429/503 can use Retry-After instead of 10s | |
| raise_on_status=False, # lets you receive the final response even if status is bad | |
| ) | |
| s = requests.Session() | |
| adapter = HTTPAdapter(max_retries=retry) | |
| s.mount("https://", adapter) | |
| s.mount("http://", adapter) | |
| return s | |
| if __name__ == "__main__": | |
| logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(name)s: %(message)s") | |
| log.setLevel(logging.INFO) | |
| session = build_session() | |
| # log of the first attempt (optional — outside Retry) | |
| url = "http://127.0.0.1:5000/test" | |
| log.info("Starting request: GET %s (attempt #1)", url) | |
| resp = session.get(url, timeout=10) | |
| log.info("Finished: status=%s", resp.status_code) |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| import logging | |
| from flask import Flask, jsonify, request | |
| app = Flask(__name__) | |
| requests_counter = 0 | |
| logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(name)s: %(message)s") | |
| log = logging.getLogger("server.flask") | |
| log.setLevel(logging.INFO) | |
| def log_request(request, action): | |
| global requests_counter | |
| log.info(f"[{requests_counter}] Request received: {request.method} {request.path} from {request.remote_addr} -> Action: {action}") | |
| @app.get("/test") | |
| def test(): | |
| global requests_counter | |
| requests_counter += 1 | |
| if requests_counter == 5: | |
| requests_counter = 0 | |
| log_request(request, '200 - OKAY') | |
| return jsonify({"status": "active"}), 200 | |
| log_request(request, '429 - Retry-After') | |
| return jsonify({"status": "try again"}), 429 | |
| @app.get("/reset") | |
| def reset(): | |
| global requests_counter | |
| requests_counter = 0 | |
| log.info(f"-> Request received: {request.method} {request.path} from {request.remote_addr}") | |
| return jsonify({"status": "reset"}), 200 | |
| if __name__ == "__main__": | |
| app.run(host="0.0.0.0", port=5000, use_reloader=True) |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment