Skip to content

backoff

Build Status PyPI Version License

Function decoration for backoff and retry

This module provides function decorators which can be used to wrap a function such that it will be retried until some condition is met. It is meant to be of use when accessing unreliable resources with the potential for intermittent failures (network resources, external APIs, etc).

Features

  • Simple decorators - Easy-to-use @backoff.on_exception and @backoff.on_predicate decorators
  • Multiple wait strategies - Exponential, fibonacci, constant, and runtime-configurable strategies
  • Flexible configuration - Control retry limits with max_time, max_tries, and custom give-up conditions
  • Event handlers - Hook into retry lifecycle with on_success, on_backoff, and on_giveup callbacks
  • Async support - Full support for asyncio coroutines
  • Type hints - Fully typed for better IDE support
  • Battle-tested - Used in production by thousands of projects

Quick Start

Install via pip:

pip install python-backoff

Basic retry on exception:

import backoff
import requests

@backoff.on_exception(backoff.expo,
                      requests.exceptions.RequestException,
                      max_time=60)
def get_url(url):
    return requests.get(url)

This will retry the function with exponential backoff whenever a RequestException is raised, giving up after 60 seconds.

Common Use Cases

API Rate Limiting

@backoff.on_predicate(
    backoff.runtime,
    predicate=lambda r: r.status_code == 429,
    value=lambda r: int(r.headers.get("Retry-After", 1)),
    jitter=None,
)
def call_api():
    return requests.get(api_url)

Database Retries

@backoff.on_exception(backoff.expo,
                      sqlalchemy.exc.OperationalError,
                      max_tries=5)
def query_database():
    return session.query(Model).all()

Polling for Results

@backoff.on_predicate(backoff.constant,
                      lambda result: result is None,
                      interval=2,
                      max_time=300)
def poll_for_result(job_id):
    return check_job_status(job_id)

Next Steps