Module slack_sdk.http_retry

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

class ConnectionErrorRetryHandler(RetryHandler):
    """RetryHandler that does retries for connectivity issues."""

    def __init__(
        self,
        max_retry_count: int = 1,
        interval_calculator: RetryIntervalCalculator = default_interval_calculator,
        error_types: List[Type[Exception]] = [
            # To cover URLError: <urlopen error [Errno 104] Connection reset by peer>
            URLError,
            ConnectionResetError,
            RemoteDisconnected,
        ],
    ):
        super().__init__(max_retry_count, interval_calculator)
        self.error_types_to_do_retries = error_types

    def _can_retry(
        self,
        *,
        state: RetryState,
        request: HttpRequest,
        response: Optional[HttpResponse] = None,
        error: Optional[Exception] = None,
    ) -> bool:
        if error is None:
            return False

        if isinstance(error, URLError):
            if response is not None:
                return False  # status 40x

        for error_type in self.error_types_to_do_retries:
            if isinstance(error, error_type):
                return True
        return False

class FixedValueRetryIntervalCalculator(RetryIntervalCalculator):
    """Retry interval calculator that uses a fixed value."""

    fixed_interval: float

    def __init__(self, fixed_internal: float = 0.5):
        """Retry interval calculator that uses a fixed value.

        Args:
            fixed_internal: The fixed interval seconds
        """
        self.fixed_interval = fixed_internal

    def calculate_sleep_duration(self, current_attempt: int) -> float:
        return self.fixed_interval

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 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 Jitter:
    """Jitter interface"""

    def recalculate(self, duration: float) -> float:
        """Recalculate the given duration.
        see also: https://aws.amazon.com/blogs/architecture/exponential-backoff-and-jitter/

        Args:
            duration: the duration in seconds

        Returns:
            A new duration that the jitter amount is added
        """
        raise NotImplementedError()

class RateLimitErrorRetryHandler(RetryHandler):
    """RetryHandler that does retries for rate limited errors."""

    def _can_retry(
        self,
        *,
        state: RetryState,
        request: HttpRequest,
        response: Optional[HttpResponse] = None,
        error: Optional[Exception] = None,
    ) -> bool:
        return response is not None and response.status_code == 429

    def prepare_for_next_attempt(
        self,
        *,
        state: RetryState,
        request: HttpRequest,
        response: Optional[HttpResponse] = None,
        error: Optional[Exception] = None,
    ) -> None:
        if response is None:
            raise error

        state.next_attempt_requested = True
        retry_after_header_name: Optional[str] = None
        for k in response.headers.keys():
            if k.lower() == "retry-after":
                retry_after_header_name = k
                break
        duration = 1
        if retry_after_header_name is None:
            # This situation usually does not arise. Just in case.
            duration += random.random()
        else:
            duration = int(response.headers.get(retry_after_header_name)[0]) + random.random()
        time.sleep(duration)
        state.increment_current_attempt()

class RetryHandler:
    """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

    def can_retry(
        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 self._can_retry(
            state=state,
            request=request,
            response=response,
            error=error,
        )

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

    def prepare_for_next_attempt(
        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)
        time.sleep(duration)
        state.increment_current_attempt()

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

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