Module `slack_sdk.http_retry.async_handler`

asyncio compatible RetryHandler interface. You can pass an array of handlers to customize retry logics in supported API clients.

Classes

class BackoffRetryIntervalCalculator (backoff_factor: float = 0.5, jitter: Optional[Jitter] = None)

Retry interval calculator that calculates in the manner of Exponential Backoff And Jitter see also: https://aws.amazon.com/blogs/architecture/exponential-backoff-and-jitter/

Retry interval calculator that calculates in the manner of Exponential Backoff And Jitter

Args

backoff_factor: The factor for the backoff interval calculation
jitter: The jitter logic implementation

Expand source code

class BackoffRetryIntervalCalculator(RetryIntervalCalculator):
    """Retry interval calculator that calculates in the manner of Exponential Backoff And Jitter
    see also: https://aws.amazon.com/blogs/architecture/exponential-backoff-and-jitter/
    """

    backoff_factor: float
    jitter: Jitter

    def __init__(self, backoff_factor: float = 0.5, jitter: Optional[Jitter] = None):
        """Retry interval calculator that calculates in the manner of Exponential Backoff And Jitter

        Args:
            backoff_factor: The factor for the backoff interval calculation
            jitter: The jitter logic implementation
        """
        self.backoff_factor = backoff_factor
        self.jitter = jitter if jitter is not None else RandomJitter()

    def calculate_sleep_duration(self, current_attempt: int) -> float:
        interval = self.backoff_factor * (2 ** (current_attempt))
        sleep_duration = self.jitter.recalculate(interval)
        return sleep_duration

Ancestors

RetryIntervalCalculator

Class variables

var backoff_factor : float
var jitter : Jitter

Inherited members

RetryIntervalCalculator:
- calculate_sleep_duration

class HttpRequest (*, method: str, url: str, headers: Dict[str, Union[str, List[str]]], body_params: Optional[Dict[str, Any]] = None, data: Optional[bytes] = None)

HTTP request representation

Expand source code

class HttpRequest:
    """HTTP request representation"""

    method: str
    url: str
    headers: Dict[str, Union[str, List[str]]]
    body_params: Optional[Dict[str, Any]]
    data: Optional[bytes]

    def __init__(
        self,
        *,
        method: str,
        url: str,
        headers: Dict[str, Union[str, List[str]]],
        body_params: Optional[Dict[str, Any]] = None,
        data: Optional[bytes] = None,
    ):
        self.method = method
        self.url = url
        self.headers = {k: v if isinstance(v, list) else [v] for k, v in headers.items()}
        self.body_params = body_params
        self.data = data

    @classmethod
    def from_urllib_http_request(cls, req: Request) -> "HttpRequest":
        return HttpRequest(
            method=req.method,
            url=req.full_url,
            headers={k: v if isinstance(v, list) else [v] for k, v in req.headers.items()},
            data=req.data,
        )

Class variables

var body_params : Optional[Dict[str, Any]]
var data : Optional[bytes]
var headers : Dict[str, Union[str, List[str]]]
var method : str
var url : str

Static methods

def from_urllib_http_request(req: urllib.request.Request) ‑> HttpRequest

class HttpResponse (*, status_code: Union[str, int], headers: Dict[str, Union[str, List[str]]], body: Optional[Dict[str, Any]] = None, data: Optional[bytes] = None)

HTTP response representation

Expand source code

class HttpResponse:
    """HTTP response representation"""

    status_code: int
    headers: Dict[str, List[str]]
    body: Optional[Dict[str, Any]]
    data: Optional[bytes]

    def __init__(
        self,
        *,
        status_code: Union[int, str],
        headers: Dict[str, Union[str, List[str]]],
        body: Optional[Dict[str, Any]] = None,
        data: Optional[bytes] = None,
    ):
        self.status_code = int(status_code)
        self.headers = {k: v if isinstance(v, list) else [v] for k, v in headers.items()}
        self.body = body
        self.data = data

Class variables

var body : Optional[Dict[str, Any]]
var data : Optional[bytes]
var headers : Dict[str, List[str]]
var status_code : int

class RetryIntervalCalculator

Retry interval calculator interface.

Expand source code

class RetryIntervalCalculator:
    """Retry interval calculator interface."""

    def calculate_sleep_duration(self, current_attempt: int) -> float:
        """Calculates an interval duration in seconds.

        Args:
            current_attempt: the number of the current attempt (zero-origin; 0 means no retries are done so far)
        Returns:
            calculated interval duration in seconds
        """
        raise NotImplementedError()

Subclasses

Methods

def calculate_sleep_duration(self, current_attempt: int) ‑> float

Calculates an interval duration in seconds.

Args

current_attempt: the number of the current attempt (zero-origin; 0 means no retries are done so far)

Returns

calculated interval duration in seconds

class RetryState (*, current_attempt: int = 0, custom_values: Optional[Dict[str, Any]] = None)

Expand source code

class RetryState:
    next_attempt_requested: bool
    current_attempt: int  # zero-origin
    custom_values: Optional[Dict[str, Any]]

    def __init__(
        self,
        *,
        current_attempt: int = 0,
        custom_values: Optional[Dict[str, Any]] = None,
    ):
        self.next_attempt_requested = False
        self.current_attempt = current_attempt
        self.custom_values = custom_values

    def increment_current_attempt(self) -> int:
        self.current_attempt += 1
        return self.current_attempt

Class variables

var current_attempt : int
var custom_values : Optional[Dict[str, Any]]
var next_attempt_requested : bool

Methods

def increment_current_attempt(self) ‑> int