Add a retrying HTTP helper shared by every remote call
One place decides what is worth another attempt: 429, 5xx and network errors, with jittered exponential backoff and Retry-After honoured. Other 4xx responses fail at once. Downloads stream to disk under the same policy, refuse a body over the size cap, and only flickr calls and photo downloads are paced; the api key is redacted from debug output. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
This commit is contained in:
2 files changed
+173
No files matched your search
@@ -272,5 +272,111 @@ def settings(args, environ=None):
|
||||
return conf
|
||||
|
||||
|
||||
# Flickr's key rides in the query string, and a debug log is a file or a
|
||||
# cron mail that outlives the run, so keep credentials out of it.
|
||||
SECRETS = re.compile(r"(?i)([?&](?:api_key|api_sig|auth_token|oauth_token)=)[^&]*")
|
||||
|
||||
def redact(url):
|
||||
return SECRETS.sub(r"\1<redacted>", url or "")
|
||||
|
||||
def retryafter(e):
|
||||
value = e.headers.get("Retry-After") if e.headers else None
|
||||
if not value:
|
||||
return None
|
||||
try:
|
||||
return max(0.0, float(value))
|
||||
except ValueError:
|
||||
pass
|
||||
try:
|
||||
when = email.utils.parsedate_to_datetime(value)
|
||||
return max(0.0, when.timestamp() - time.time())
|
||||
except Exception:
|
||||
return None
|
||||
|
||||
# Exponential backoff with jitter, so many photos tripping a limit at once do
|
||||
# not then retry in lockstep.
|
||||
def backoff(attempt):
|
||||
return min(BACKOFF_BASE * (2 ** (attempt - 1)), BACKOFF_CAP) * (0.5 + random.random())
|
||||
|
||||
#
|
||||
# Fetch a url, retrying on rate limits, server errors and transient network
|
||||
# failures. consume() is handed the open response so a body can be read into
|
||||
# memory or streamed to disk under one retry policy. 429 and 5xx are worth
|
||||
# another go; any other 4xx is our fault and will not improve. Only flickr
|
||||
# needs pacing, and only flickr's limit is worth slowing the run for.
|
||||
#
|
||||
def retrying(url, context, consume, headers=None, data=None, method="GET", pace=False):
|
||||
sent = {"User-Agent": USER_AGENT}
|
||||
sent.update(headers or {})
|
||||
for attempt in range(1, ATTEMPTS + 1):
|
||||
if pace and PACE:
|
||||
SLEEP(PACE)
|
||||
debug("%s: %s %s%s" % (context, method, redact(url),
|
||||
" (attempt %d)" % attempt if attempt > 1 else ""))
|
||||
request = urllib.request.Request(url, data=data, method=method, headers=sent)
|
||||
try:
|
||||
response = OPEN(request, timeout=HTTP_TIMEOUT)
|
||||
try:
|
||||
return consume(response)
|
||||
finally:
|
||||
response.close()
|
||||
except BackupError:
|
||||
raise # consume() rejected the body; retrying will not help
|
||||
except urllib.error.HTTPError as e:
|
||||
if e.code != 429 and e.code < 500:
|
||||
detail = ""
|
||||
try:
|
||||
detail = ": " + e.read().decode("utf-8", "replace").strip()[:200]
|
||||
except Exception:
|
||||
pass
|
||||
e.close()
|
||||
raise HTTPFailure("%s: HTTP %s %s%s" % (context, e.code, e.reason, detail),
|
||||
code=e.code)
|
||||
wait = retryafter(e)
|
||||
reason = "HTTP %s %s" % (e.code, e.reason)
|
||||
code = e.code
|
||||
e.close()
|
||||
except Exception as e:
|
||||
wait = None
|
||||
reason = "%s: %s" % (type(e).__name__, e)
|
||||
code = None
|
||||
|
||||
if attempt == ATTEMPTS:
|
||||
raise HTTPFailure("%s: giving up after %d attempts (%s)"
|
||||
% (context, ATTEMPTS, reason), code=code)
|
||||
if wait is None:
|
||||
wait = backoff(attempt)
|
||||
STATS["retries"] += 1
|
||||
warn("%s: %s - retrying in %.0fs (attempt %d of %d)"
|
||||
% (context, reason, wait, attempt + 1, ATTEMPTS))
|
||||
SLEEP(wait)
|
||||
|
||||
def fetch(url, context, headers=None, data=None, method="GET", pace=False):
|
||||
return retrying(url, context, lambda r: r.read(), headers, data, method, pace)
|
||||
|
||||
#
|
||||
# Stream a url to a file, written as it arrives. Returns the Content-Type the
|
||||
# server reported and the byte count. The caller passes a .part path and
|
||||
# renames it once the sidecar is written too. A body past the cap is a
|
||||
# redirect to something that is not a photo, or a server gone wrong; stop
|
||||
# rather than fill the disk.
|
||||
#
|
||||
def download(url, context, path, pace=False):
|
||||
def consume(response):
|
||||
written = 0
|
||||
with open(path, "wb") as fh:
|
||||
while True:
|
||||
chunk = response.read(DOWNLOAD_CHUNK)
|
||||
if not chunk:
|
||||
break
|
||||
fh.write(chunk)
|
||||
written += len(chunk)
|
||||
if written > DOWNLOAD_CAP:
|
||||
raise BackupError("%s: larger than %d bytes, refusing it"
|
||||
% (context, DOWNLOAD_CAP))
|
||||
return (response.headers.get("Content-Type", ""), written)
|
||||
return retrying(url, context, consume, pace=pace)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(0)
|
||||
@@ -236,3 +236,70 @@ class Settings(TestCase):
|
||||
def test_tilde_expanded_in_paths(self):
|
||||
path = self.write("s.yml", "directory: ~/photos\n")
|
||||
self.assertEqual(rb.loadsettings(path)["directory"], os.path.expanduser("~/photos"))
|
||||
|
||||
|
||||
class Retrying(TestCase):
|
||||
def test_success_returns_body(self):
|
||||
rb.OPEN = FakeOpen(FakeResponse(b"ok"))
|
||||
self.assertEqual(rb.fetch("http://t/x", "ctx"), b"ok")
|
||||
self.assertEqual(rb.OPEN.requests[0].get_header("User-agent"), rb.USER_AGENT)
|
||||
|
||||
def test_404_is_not_retried(self):
|
||||
rb.OPEN = FakeOpen(http_error(404, b"gone"))
|
||||
with self.assertRaises(rb.HTTPFailure) as cm:
|
||||
rb.fetch("http://t/x", "ctx")
|
||||
self.assertEqual(cm.exception.code, 404)
|
||||
self.assertIn("gone", str(cm.exception))
|
||||
self.assertEqual(len(rb.OPEN.requests), 1)
|
||||
|
||||
def test_5xx_and_429_retry_then_succeed(self):
|
||||
rb.OPEN = FakeOpen(http_error(503), http_error(429), FakeResponse(b"ok"))
|
||||
self.assertEqual(rb.fetch("http://t/x", "ctx"), b"ok")
|
||||
self.assertEqual(len(rb.OPEN.requests), 3)
|
||||
self.assertEqual(rb.STATS["retries"], 2)
|
||||
|
||||
def test_connection_errors_retry(self):
|
||||
rb.OPEN = FakeOpen(OSError("reset"), FakeResponse(b"ok"))
|
||||
self.assertEqual(rb.fetch("http://t/x", "ctx"), b"ok")
|
||||
|
||||
def test_gives_up_after_attempts(self):
|
||||
rb.ATTEMPTS = 3
|
||||
rb.OPEN = FakeOpen(http_error(500), http_error(500), http_error(500))
|
||||
with self.assertRaises(rb.HTTPFailure) as cm:
|
||||
rb.fetch("http://t/x", "ctx")
|
||||
self.assertIn("3 attempts", str(cm.exception))
|
||||
|
||||
def test_retry_after_header_is_honoured(self):
|
||||
waits = []
|
||||
rb.SLEEP = waits.append
|
||||
rb.OPEN = FakeOpen(http_error(429, headers={"Retry-After": "12"}), FakeResponse(b"ok"))
|
||||
rb.fetch("http://t/x", "ctx")
|
||||
self.assertIn(12.0, waits)
|
||||
|
||||
def test_backoff_grows_and_caps(self):
|
||||
rb.BACKOFF_BASE = 5
|
||||
self.assertLessEqual(rb.backoff(1), 7.5)
|
||||
self.assertGreaterEqual(rb.backoff(1), 2.5)
|
||||
self.assertLessEqual(rb.backoff(20), rb.BACKOFF_CAP * 1.5)
|
||||
|
||||
def test_post_sends_body_and_headers(self):
|
||||
rb.OPEN = FakeOpen(FakeResponse(b"{}"))
|
||||
rb.fetch("http://t/x", "ctx", headers={"x-api-key": "k"}, data=b"body", method="POST")
|
||||
req = rb.OPEN.requests[0]
|
||||
self.assertEqual(req.get_method(), "POST")
|
||||
self.assertEqual(req.data, b"body")
|
||||
self.assertEqual(req.get_header("X-api-key"), "k")
|
||||
|
||||
def test_download_streams_to_path(self):
|
||||
rb.OPEN = FakeOpen(FakeResponse(b"x" * 1000, headers={"Content-Type": "image/jpeg"}))
|
||||
path = os.path.join(self.dir, "p.part")
|
||||
self.assertEqual(rb.download("http://t/p.jpg", "ctx", path), ("image/jpeg", 1000))
|
||||
self.assertEqual(os.path.getsize(path), 1000)
|
||||
def test_download_refuses_an_oversized_body(self):
|
||||
rb.DOWNLOAD_CAP = 10
|
||||
rb.OPEN = FakeOpen(FakeResponse(b"x" * 100))
|
||||
path = os.path.join(self.dir, "p.part")
|
||||
with self.assertRaises(rb.BackupError) as cm:
|
||||
rb.download("http://t/p.jpg", "ctx", path)
|
||||
self.assertIn("10", str(cm.exception))
|
||||
self.assertEqual(len(rb.OPEN.requests), 1) # not a retryable failure
|
||||
Reference in new issue
Block a user