CtrlK
BlogDocsLog inGet started
Tessl Logo

tessl/pypi-aliyun-python-sdk-core

The core module of Aliyun Python SDK providing authentication, request handling, and API communication for Alibaba Cloud services.

—

Pending

Quality

Pending

Does it follow best practices?

Impact

Pending

No eval scenarios have been run

Overview
Eval results
Files

retry-backoff.mddocs/

Retry & Backoff

Flexible retry mechanisms with configurable retry conditions and backoff strategies for handling transient failures, rate limiting, and network issues. The SDK provides comprehensive retry policies to improve reliability and resilience.

Capabilities

Retry Policy

Main retry policy class that combines retry conditions and backoff strategies to define complete retry behavior.

class RetryPolicy(RetryCondition, BackoffStrategy):
    def __init__(self, retry_condition, backoff_strategy):
        """
        Initialize retry policy with condition and backoff strategy.
        
        Parameters:
        - retry_condition (RetryCondition): Condition determining when to retry
        - backoff_strategy (BackoffStrategy): Strategy for delay between retries
        """
    
    def should_retry(self, retry_policy_context):
        """
        Determine if request should be retried based on the condition.
        
        Parameters:
        - retry_policy_context (RetryPolicyContext): Context with attempt info
        
        Returns:
        bool: True if request should be retried
        """
    
    def compute_delay_before_next_retry(self, retry_policy_context):
        """
        Calculate delay before next retry attempt.
        
        Parameters:
        - retry_policy_context (RetryPolicyContext): Context with attempt info
        
        Returns:
        float: Delay in seconds before next retry
        """

Retry Conditions

Base class and implementations for different retry conditions that determine when requests should be retried.

class RetryCondition:
    def should_retry(self, retry_policy_context):
        """
        Abstract method to determine if retry should be attempted.
        
        Parameters:
        - retry_policy_context (RetryPolicyContext): Context with attempt info
        
        Returns:
        bool: True if retry should be attempted
        """

class NoRetryCondition(RetryCondition):
    """Condition that never retries."""

class MaxRetryTimesCondition(RetryCondition):
    def __init__(self, max_retry_times):
        """
        Retry condition based on maximum number of attempts.
        
        Parameters:
        - max_retry_times (int): Maximum number of retry attempts
        """

class RetryOnExceptionCondition(RetryCondition):
    def __init__(self, retryable_exceptions):
        """
        Retry condition based on specific exception types.
        
        Parameters:
        - retryable_exceptions (list): List of exception types that should trigger retry
        """

class RetryOnHttpStatusCondition(RetryCondition):
    def __init__(self, retryable_status_codes):
        """
        Retry condition based on HTTP status codes.
        
        Parameters:
        - retryable_status_codes (list): List of HTTP status codes that should trigger retry
        """

class RetryOnApiCondition(RetryCondition):
    def __init__(self, retryable_error_codes):
        """
        Retry condition based on API error codes.
        
        Parameters:
        - retryable_error_codes (list): List of API error codes that should trigger retry
        """

Backoff Strategies

Different backoff strategies for controlling the delay between retry attempts.

class BackoffStrategy:
    def compute_delay_before_next_retry(self, retry_policy_context):
        """
        Abstract method to compute delay before next retry.
        
        Parameters:
        - retry_policy_context (RetryPolicyContext): Context with attempt info
        
        Returns:
        float: Delay in seconds
        """

class FixedDelayStrategy(BackoffStrategy):
    def __init__(self, fixed_delay):
        """
        Fixed delay strategy with constant interval between retries.
        
        Parameters:
        - fixed_delay (float): Fixed delay in seconds between retries
        """

class NoDelayStrategy(FixedDelayStrategy):
    """No delay strategy for immediate retries."""

class ExponentialBackoffStrategy(BackoffStrategy):
    def __init__(self, base_delay=1.0, max_delay=60.0, multiplier=2.0):
        """
        Exponential backoff strategy with increasing delays.
        
        Parameters:
        - base_delay (float): Initial delay in seconds, defaults to 1.0
        - max_delay (float): Maximum delay in seconds, defaults to 60.0
        - multiplier (float): Multiplier for each retry, defaults to 2.0
        """

class JitteredExponentialBackoffStrategy(ExponentialBackoffStrategy):
    def __init__(self, base_delay=1.0, max_delay=60.0, multiplier=2.0):
        """
        Exponential backoff with random jitter to avoid thundering herd.
        
        Parameters:
        - base_delay (float): Initial delay in seconds
        - max_delay (float): Maximum delay in seconds
        - multiplier (float): Multiplier for each retry
        """

Retry Policy Context

Context object that tracks retry attempt information and provides data to retry conditions and backoff strategies.

class RetryPolicyContext:
    def __init__(self):
        """Initialize retry policy context for tracking attempt information."""
    
    def get_retrials_attempted(self):
        """
        Get the number of retry attempts made so far.
        
        Returns:
        int: Number of retries attempted
        """
    
    def get_exception(self):
        """
        Get the exception that triggered the retry consideration.
        
        Returns:
        Exception: The exception from the failed attempt
        """
    
    def get_original_request(self):
        """
        Get the original request being retried.
        
        Returns:
        AcsRequest: The original request object
        """

Composite Retry Conditions

Advanced retry conditions that combine multiple conditions with logical operators.

class AndRetryCondition(RetryCondition):
    def __init__(self, *conditions):
        """
        Retry condition that requires ALL conditions to be true.
        
        Parameters:
        - conditions: Variable number of RetryCondition objects
        """

class OrRetryCondition(RetryCondition):
    def __init__(self, *conditions):
        """
        Retry condition that requires ANY condition to be true.
        
        Parameters:
        - conditions: Variable number of RetryCondition objects
        """

class MixedRetryCondition(RetryCondition):
    """Complex retry condition combining multiple criteria."""

class DefaultConfigRetryCondition(MixedRetryCondition):
    """Default retry condition with reasonable settings for most use cases."""

Usage Examples

Basic Retry Configuration

from aliyunsdkcore.client import AcsClient
from aliyunsdkcore.retry.retry_policy import RetryPolicy
from aliyunsdkcore.retry.retry_condition import MaxRetryTimesCondition
from aliyunsdkcore.retry.backoff_strategy import FixedDelayStrategy

# Create retry policy with fixed delay
retry_condition = MaxRetryTimesCondition(max_retry_times=3)
backoff_strategy = FixedDelayStrategy(fixed_delay=2.0)
retry_policy = RetryPolicy(retry_condition, backoff_strategy)

# Configure client with retry policy
client = AcsClient(
    ak="your-ak",
    secret="your-secret",
    region_id="cn-hangzhou",
    auto_retry=True,
    max_retry_time=3
)

Exponential Backoff with Jitter

from aliyunsdkcore.retry.backoff_strategy import JitteredExponentialBackoffStrategy
from aliyunsdkcore.retry.retry_condition import MaxRetryTimesCondition

# Create exponential backoff with jitter
retry_condition = MaxRetryTimesCondition(max_retry_times=5)
backoff_strategy = JitteredExponentialBackoffStrategy(
    base_delay=1.0,
    max_delay=30.0,
    multiplier=2.0
)
retry_policy = RetryPolicy(retry_condition, backoff_strategy)

client = AcsClient(
    ak="your-ak",
    secret="your-secret",
    region_id="cn-hangzhou",
    retry_policy=retry_policy
)

Exception-Based Retry

from aliyunsdkcore.retry.retry_condition import RetryOnExceptionCondition
from aliyunsdkcore.acs_exception.exceptions import ClientException, ServerException

# Retry on specific exceptions
retryable_exceptions = [
    ClientException,  # Network or client errors
    # Add other exception types as needed
]

retry_condition = RetryOnExceptionCondition(retryable_exceptions)
backoff_strategy = ExponentialBackoffStrategy(base_delay=2.0, max_delay=60.0)
retry_policy = RetryPolicy(retry_condition, backoff_strategy)

HTTP Status Code Based Retry

from aliyunsdkcore.retry.retry_condition import RetryOnHttpStatusCondition

# Retry on specific HTTP status codes
retryable_status_codes = [
    429,  # Too Many Requests
    500,  # Internal Server Error
    502,  # Bad Gateway
    503,  # Service Unavailable
    504,  # Gateway Timeout
]

retry_condition = RetryOnHttpStatusCondition(retryable_status_codes)
backoff_strategy = JitteredExponentialBackoffStrategy()
retry_policy = RetryPolicy(retry_condition, backoff_strategy)

API Error Code Based Retry

from aliyunsdkcore.retry.retry_condition import RetryOnApiCondition

# Retry on specific API error codes
retryable_error_codes = [
    "Throttling",
    "ServiceUnavailable", 
    "InternalError",
    "RequestTimeout"
]

retry_condition = RetryOnApiCondition(retryable_error_codes)
backoff_strategy = ExponentialBackoffStrategy()
retry_policy = RetryPolicy(retry_condition, backoff_strategy)

Complex Retry Conditions

from aliyunsdkcore.retry.retry_condition import AndRetryCondition, OrRetryCondition

# Combine multiple conditions with AND logic
max_retries = MaxRetryTimesCondition(max_retry_times=3)
http_status = RetryOnHttpStatusCondition([500, 502, 503])
and_condition = AndRetryCondition(max_retries, http_status)

# Combine multiple conditions with OR logic  
exception_condition = RetryOnExceptionCondition([ClientException])
api_condition = RetryOnApiCondition(["Throttling"])
or_condition = OrRetryCondition(exception_condition, api_condition)

# Use complex condition
retry_policy = RetryPolicy(or_condition, ExponentialBackoffStrategy())

Custom Retry Implementation

import time
import random
from aliyunsdkcore.acs_exception.exceptions import ClientException, ServerException

def custom_retry_request(client, request, max_retries=3):
    """
    Custom retry implementation with exponential backoff and jitter.
    """
    
    for attempt in range(max_retries + 1):
        try:
            response = client.do_action_with_exception(request)
            return response.decode('utf-8')
            
        except (ClientException, ServerException) as e:
            if attempt == max_retries:
                raise  # Re-raise on final attempt
            
            # Determine if we should retry
            should_retry = False
            
            if isinstance(e, ClientException):
                # Retry on network/timeout errors
                if e.get_error_code() in ["SDK.NetworkError", "SDK.TimeoutError"]:
                    should_retry = True
                    
            elif isinstance(e, ServerException):
                # Retry on throttling and 5xx errors
                if (e.get_error_code() == "Throttling" or 
                    (e.get_http_status() and e.get_http_status() >= 500)):
                    should_retry = True
            
            if should_retry:
                # Exponential backoff with jitter
                base_delay = 2.0
                max_delay = 60.0
                backoff = min(base_delay * (2 ** attempt), max_delay)
                jitter = random.uniform(0, backoff * 0.1)  # 10% jitter
                delay = backoff + jitter
                
                print(f"Retrying in {delay:.2f} seconds (attempt {attempt + 1})")
                time.sleep(delay)
            else:
                raise  # Don't retry non-retryable errors

# Usage
try:
    result = custom_retry_request(client, request, max_retries=5)
    print("Success:", result)
except Exception as e:
    print(f"Failed after retries: {e}")

Default Retry Configuration

from aliyunsdkcore.retry.retry_condition import DefaultConfigRetryCondition
from aliyunsdkcore.retry.backoff_strategy import DefaultMixedBackoffStrategy

# Use default configurations for common scenarios
retry_condition = DefaultConfigRetryCondition()
backoff_strategy = DefaultMixedBackoffStrategy()
retry_policy = RetryPolicy(retry_condition, backoff_strategy)

# Apply to client
client = AcsClient(
    ak="your-ak",
    secret="your-secret",
    region_id="cn-hangzhou",
    auto_retry=True,
    retry_policy=retry_policy
)

Monitoring Retry Attempts

import logging
from aliyunsdkcore.retry.retry_policy_context import RetryPolicyContext

logger = logging.getLogger(__name__)

class LoggingRetryPolicy(RetryPolicy):
    """Custom retry policy with logging."""
    
    def should_retry(self, retry_policy_context):
        should_retry = super().should_retry(retry_policy_context)
        
        if should_retry:
            attempt = retry_policy_context.get_retrials_attempted()
            exception = retry_policy_context.get_exception()
            logger.info(f"Retrying request (attempt {attempt + 1}): {exception}")
        
        return should_retry
    
    def compute_delay_before_next_retry(self, retry_policy_context):
        delay = super().compute_delay_before_next_retry(retry_policy_context)
        logger.info(f"Waiting {delay:.2f} seconds before retry")
        return delay

# Use logging retry policy
retry_policy = LoggingRetryPolicy(
    MaxRetryTimesCondition(3),
    ExponentialBackoffStrategy()
)

Install with Tessl CLI

npx tessl i tessl/pypi-aliyun-python-sdk-core

docs

authentication.md

client-management.md

exception-handling.md

index.md

request-types.md

retry-backoff.md

tile.json