tessl/pypi-tencentcloud-sdk-python-common

Common utilities and core components for Tencent Cloud SDK for Python, providing credential management, HTTP clients, authentication, error handling, and foundational classes required by all Tencent Cloud service SDKs

Overview

Eval results

Files

Resilience and Reliability

Name: tessl/pypi-tencentcloud-sdk-python-common
Author: tessl

Circuit breaker implementation for regional failover and retry mechanisms with configurable backoff strategies. Provides automatic failure detection and recovery for improved reliability in distributed cloud environments.

Capabilities

Circuit Breaker Implementation

Circuit breaker pattern implementation to prevent cascading failures and enable automatic failover to backup regions.

class CircuitBreaker:
    def __init__(self, breaker_setting):
        """
        Create circuit breaker.
        
        Args:
            breaker_setting: RegionBreakerProfile configuration object
        """
        
    def before_requests(self) -> tuple[int, bool]:
        """
        Check circuit breaker state before making request.
        
        Returns:
            tuple: (generation_id, use_backup_region)
                generation_id (int): Current generation for tracking
                use_backup_region (bool): Whether to use backup region
        """
        
    def after_requests(self, before: int, success: bool) -> None:
        """
        Record request result after completion.
        
        Args:
            before (int): Generation ID from before_requests()
            success (bool): Whether the request succeeded
        """

Request Counter

Counter for tracking request statistics used by the circuit breaker.

class Counter:
    def __init__(self):
        """Create request counter."""
        
    def on_success(self) -> None:
        """Record successful request."""
        
    def on_failure(self) -> None:
        """Record failed request."""
        
    def clear(self) -> None:
        """Reset all counters."""
        
    def get_failure_rate(self) -> float:
        """
        Calculate failure rate.
        
        Returns:
            float: Failure rate (0.0 to 1.0)
        """

Circuit Breaker States

STATE_CLOSED: int = 0      # Circuit breaker is closed (normal operation)
STATE_HALF_OPEN: int = 1   # Circuit breaker is half-open (testing recovery)
STATE_OPEN: int = 2        # Circuit breaker is open (using backup region)

Retry Mechanisms

No-operation retry policy that performs no retries.

class NoopRetryer:
    def send_request(self, fn):
        """
        Execute function without retry.
        
        Args:
            fn: Function to execute
            
        Returns:
            Result of function execution
        """

Standard retry policy with exponential backoff and configurable retry conditions.

class StandardRetryer:
    def __init__(self, max_attempts: int = 3, backoff_fn = None, logger = None):
        """
        Create standard retryer.
        
        Args:
            max_attempts (int): Maximum number of attempts. Default: 3
            backoff_fn: Backoff function taking attempt number, returns sleep seconds
            logger: Logger instance for retry logging
        """
        
    def send_request(self, fn):
        """
        Execute function with retry logic.
        
        Args:
            fn: Function to execute
            
        Returns:
            Result of successful function execution
            
        Raises:
            TencentCloudSDKException: If all retry attempts fail
        """
        
    @staticmethod
    def should_retry(resp, err) -> bool:
        """
        Determine if request should be retried.
        
        Args:
            resp: Response object (if any)
            err: Exception object (if any)
            
        Returns:
            bool: True if request should be retried
        """
        
    @staticmethod
    def backoff(n: int) -> int:
        """
        Default exponential backoff function.
        
        Args:
            n (int): Attempt number (0-based)
            
        Returns:
            int: Sleep time in seconds (2^n)
        """
        
    def on_retry(self, n: int, sleep: int, resp, err) -> None:
        """
        Called before each retry attempt.
        
        Args:
            n (int): Attempt number
            sleep (int): Sleep time before retry
            resp: Response object (if any)
            err: Exception object (if any)
        """

Usage Examples

Basic Circuit Breaker

from tencentcloud.common.circuit_breaker import CircuitBreaker
from tencentcloud.common.profile.client_profile import RegionBreakerProfile

# Create breaker configuration
breaker_config = RegionBreakerProfile(
    backup_endpoint="ap-shanghai.tencentcloudapi.com",
    max_fail_num=5,
    max_fail_percent=0.75,
    window_interval=300,  # 5 minutes
    timeout=60,          # 1 minute
    max_requests=5
)

# Create circuit breaker
breaker = CircuitBreaker(breaker_config)

# Use circuit breaker around requests
def make_request():
    generation, use_backup = breaker.before_requests()
    
    try:
        # Make API request (use backup endpoint if use_backup is True)
        if use_backup:
            endpoint = "ap-shanghai.tencentcloudapi.com"
        else:
            endpoint = "cvm.tencentcloudapi.com"
            
        # Simulate API call
        response = api_call(endpoint)
        breaker.after_requests(generation, True)  # Success
        return response
        
    except Exception as e:
        breaker.after_requests(generation, False)  # Failure
        raise e

Standard Retry Configuration

from tencentcloud.common.retry import StandardRetryer
import logging

# Create logger for retry events
logger = logging.getLogger("retry")
logger.setLevel(logging.DEBUG)

# Create retryer with custom backoff
def custom_backoff(attempt):
    """Custom backoff: 1s, 2s, 4s, then cap at 10s"""
    return min(2 ** attempt, 10)

retryer = StandardRetryer(
    max_attempts=5,
    backoff_fn=custom_backoff,
    logger=logger
)

# Use retryer
def api_call():
    # Simulate API call that might fail
    import random
    if random.random() < 0.7:  # 70% chance of failure
        raise TencentCloudSDKException("RequestLimitExceeded", "Rate limited")
    return {"success": True}

try:
    result = retryer.send_request(api_call)
    print("Success:", result)
except Exception as e:
    print("All retries failed:", e)

No Retry Policy

from tencentcloud.common.retry import NoopRetryer

# Create no-retry policy
retryer = NoopRetryer()

# Use no-retry retryer (executes once only)
def api_call():
    return {"data": "response"}

result = retryer.send_request(api_call)
print(result)

Integration with Client Profile

from tencentcloud.common.credential import Credential
from tencentcloud.common.common_client import CommonClient
from tencentcloud.common.profile.client_profile import ClientProfile, RegionBreakerProfile
from tencentcloud.common.retry import StandardRetryer
import logging

# Setup logging
logging.basicConfig(level=logging.DEBUG)
logger = logging.getLogger("sdk_retry")

# Create credentials
cred = Credential("your-secret-id", "your-secret-key")

# Configure circuit breaker
breaker_profile = RegionBreakerProfile(
    backup_endpoint="ap-shanghai.tencentcloudapi.com",
    max_fail_num=3,
    max_fail_percent=0.6,
    window_interval=180,
    timeout=30
)

# Configure retry policy
retryer = StandardRetryer(
    max_attempts=4,
    backoff_fn=lambda n: min(1.5 ** n, 30),  # 1.5s, 2.25s, 3.4s, 30s
    logger=logger
)

# Create client profile with resilience features
client_profile = ClientProfile()
client_profile.disable_region_breaker = False
client_profile.region_breaker_profile = breaker_profile
client_profile.retryer = retryer

# Create client
client = CommonClient("cvm", "2017-03-12", cred, "ap-guangzhou", client_profile)

# API calls will automatically use circuit breaker and retry
try:
    response = client.call_json("DescribeInstances", {"Limit": 10})
    print("Success:", len(response["Response"]["InstanceSet"]))
except Exception as e:
    print("Final failure:", e)

Custom Retry Conditions

from tencentcloud.common.retry import StandardRetryer
from tencentcloud.common.exception.tencent_cloud_sdk_exception import TencentCloudSDKException

class CustomRetryer(StandardRetryer):
    @staticmethod
    def should_retry(resp, err):
        """Custom retry logic"""
        if not err:
            return False
            
        if isinstance(err, TencentCloudSDKException):
            # Retry on network errors and rate limiting
            retryable_codes = [
                "ClientNetworkError",
                "ServerNetworkError", 
                "RequestLimitExceeded",
                "InternalError"
            ]
            return err.get_code() in retryable_codes
            
        return False

# Use custom retryer
custom_retryer = CustomRetryer(max_attempts=5)

def api_call():
    # Your API call logic
    pass

result = custom_retryer.send_request(api_call)

Monitoring Circuit Breaker State

from tencentcloud.common.circuit_breaker import CircuitBreaker, STATE_CLOSED, STATE_HALF_OPEN, STATE_OPEN

class MonitoredCircuitBreaker(CircuitBreaker):
    def before_requests(self):
        generation, use_backup = super().before_requests()
        
        # Log state changes
        state_names = {
            STATE_CLOSED: "CLOSED",
            STATE_HALF_OPEN: "HALF_OPEN", 
            STATE_OPEN: "OPEN"
        }
        
        print(f"Circuit breaker state: {state_names[self.state]}")
        if use_backup:
            print("Using backup region due to circuit breaker")
            
        return generation, use_backup

# Use monitored circuit breaker
breaker = MonitoredCircuitBreaker(breaker_config)

Install with Tessl CLI

npx tessl i tessl/pypi-tencentcloud-sdk-python-common

docs

client-foundation.md

configuration-profiles.md

credential-management.md

http-infrastructure.md

index.md

resilience-reliability.md

security-authentication.md

tile.json