Python requests sessions with retries
A quick one. Python’s requests library is awesome. One thing that was a little confusing to me was adding retries (and even caching) to requests sessions. So here’s some code I worked out for that:
import datetime
import requests
import requests_cache
from requests.adapters import HTTPAdapter
import urllib3
"""
This section of code handles creating a requests Session
object with retry logic built in
We create two objects:
CACHELESS_HTTP and HTTP.
CACHELESS_HTTP is a normal Session object with a fairly aggressive retry strategy:
7 total retries, 5 of which will perform backoff
based on docs, we should see the following retry timings:
Try 2: immediately
Try 3: 12 seconds
Try 4: 24 seconds
Try 5: 36 seconds
Try 6: 48 seconds
Try 7: 60 seconds
HTTP adds a local cache (using the requests_cache library) to a Session object:
5 total retries, 3 of which will perform backoff
based on docs, we should see the following retry timings:
Try 2: immediately
Try 3: 12 seconds
Try 4: 24 seconds
Try 5: 36 seconds
You can see more details about urllib3 retries here:
https://urllib3.readthedocs.io/en/stable/reference/urllib3.util.html#urllib3.util.Retry
"""
# set status list and method list once so updates populate well
_STATUS_LIST = [413, 429, 500, 502, 503, 504]
# yes, POST isn't idempotent, but we may actually want to retry POSTs, so...
_METHOD_LIST = ["DELETE", "GET", "HEAD", "OPTIONS", "PUT", "TRACE", "POST"]
_retry_strategy = urllib3.Retry(
total=7,
status_forcelist=_STATUS_LIST,
allowed_methods=_METHOD_LIST,
backoff_factor=1,
)
_adapter = HTTPAdapter(max_retries=_retry_strategy)
CACHELESS_HTTP = requests.Session()
CACHELESS_HTTP.mount("https://", _adapter)
# if we need non-encrypted connections, un-comment
# CACHELESS_HTTP.mount("http://", _adapter)
# different strategy for potentially cached requests
_cached_retry_strategy = urllib3.Retry(
total=5,
status_forcelist=_STATUS_LIST,
allowed_methods=_METHOD_LIST,
backoff_factor=1,
)
_cached_adapter = HTTPAdapter(max_retries=_cached_retry_strategy)
"""
Use a caching instance to try to reduce frequent
queries to systems.
If we are running multiple instances of our service
this cache will require some way to be shared or
it will be far less useful than you would prefer
"""
HTTP = requests_cache.CachedSession(
"cached_session",
cache_control=True,
# default expiration of 1 hour
expire_after=datetime.timedelta(hours=1),
allowable_methods=["GET", "POST", "PUT"],
match_headers=True,
stale_if_error=True,
)
HTTP.mount("https://", _cached_adapter)
# if we need non-encrypted connections, un-comment
# HTTP.mount("http://", _cached_adapter)