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.

Expand source code

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

import asyncio
from typing import Optional

from slack_sdk.http_retry.state import RetryState
from slack_sdk.http_retry.request import HttpRequest
from slack_sdk.http_retry.response import HttpResponse
from slack_sdk.http_retry.interval_calculator import RetryIntervalCalculator
from slack_sdk.http_retry.builtin_interval_calculators import (
    BackoffRetryIntervalCalculator,
)

default_interval_calculator = BackoffRetryIntervalCalculator()


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

    max_retry_count: int
    interval_calculator: RetryIntervalCalculator

    def __init__(
        self,
        max_retry_count: int = 1,
        interval_calculator: RetryIntervalCalculator = default_interval_calculator,
    ):
        """RetryHandler interface.

        Args:
            max_retry_count: The maximum times to do retries
            interval_calculator: Pass an interval calculator for customizing the logic
        """
        self.max_retry_count = max_retry_count
        self.interval_calculator = interval_calculator

    async def can_retry_async(
        self,
        *,
        state: RetryState,
        request: HttpRequest,
        response: Optional[HttpResponse] = None,
        error: Optional[Exception] = None,
    ) -> bool:
        if state.current_attempt >= self.max_retry_count:
            return False
        return await self._can_retry_async(
            state=state,
            request=request,
            response=response,
            error=error,
        )

    async def _can_retry_async(
        self,
        *,
        state: RetryState,
        request: HttpRequest,
        response: Optional[HttpResponse] = None,
        error: Optional[Exception] = None,
    ) -> bool:
        raise NotImplementedError()

    async def prepare_for_next_attempt_async(
        self,
        *,
        state: RetryState,
        request: HttpRequest,
        response: Optional[HttpResponse] = None,
        error: Optional[Exception] = None,
    ) -> None:
        state.next_attempt_requested = True
        duration = self.interval_calculator.calculate_sleep_duration(state.current_attempt)
        await asyncio.sleep(duration)
        state.increment_current_attempt()


__all__ = [
    "RetryState",
    "HttpRequest",
    "HttpResponse",
    "RetryIntervalCalculator",
    "BackoffRetryIntervalCalculator",
    "default_interval_calculator",
]

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

Expand source code

@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 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

Expand source code

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()

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

Expand source code

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