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
Comprehensive HTTP client implementation with connection pooling, proxy support, pre-connected pools for performance optimization, and custom certificate handling. Supports both HTTP and HTTPS with configurable timeouts, keep-alive connections, and advanced connection management features.
Main HTTP request handler with connection pooling, proxy support, and configurable timeouts.
class ApiRequest:
def __init__(self, host: str, req_timeout: int = 60, debug: bool = False,
proxy: str = None, is_http: bool = False, certification = None,
pre_conn_pool_size: int = 0):
"""
Create API request handler.
Args:
host (str): Target host for requests
req_timeout (int): Request timeout in seconds (default: 60)
debug (bool): Enable debug logging (default: False)
proxy (str, optional): Proxy URL
is_http (bool): Use HTTP instead of HTTPS (default: False)
certification: Certificate path or False to disable verification
pre_conn_pool_size (int): Pre-connected pool size (default: 0)
"""
def set_req_timeout(self, req_timeout: int) -> None:
"""
Set request timeout.
Args:
req_timeout (int): Timeout in seconds
"""
def is_keep_alive(self) -> bool:
"""
Check if keep-alive is enabled.
Returns:
bool: True if keep-alive is enabled
"""
def set_keep_alive(self, flag: bool = True) -> None:
"""
Enable or disable keep-alive connections.
Args:
flag (bool): Enable keep-alive (default: True)
"""
def set_debug(self, debug: bool) -> None:
"""
Enable or disable debug logging.
Args:
debug (bool): Enable debug logging
"""
def send_request(self, req_inter: RequestInternal):
"""
Send HTTP request.
Args:
req_inter (RequestInternal): Request object
Returns:
Response: HTTP response object
Raises:
TencentCloudSDKException: On network errors or invalid methods
"""HTTP connection management with proxy support and certificate handling.
class ProxyConnection:
def __init__(self, host: str, timeout: int = 60, proxy: str = None,
certification = None, is_http: bool = False,
pre_conn_pool_size: int = 0):
"""
Create proxy connection handler.
Args:
host (str): Target host
timeout (int): Connection timeout (default: 60)
proxy (str, optional): Proxy URL
certification: Certificate path or None for default
is_http (bool): Use HTTP instead of HTTPS (default: False)
pre_conn_pool_size (int): Pre-connected pool size (default: 0)
"""
def request(self, method: str, url: str, body = None, headers: dict = None):
"""
Make HTTP request.
Args:
method (str): HTTP method (GET, POST, etc.)
url (str): Request URL
body: Request body
headers (dict, optional): Request headers
Returns:
Response: HTTP response object
"""Internal request object for building HTTP requests.
class RequestInternal:
def __init__(self, host: str = "", method: str = "", uri: str = "",
header: dict = None, data: str = ""):
"""
Create internal request representation.
Args:
host (str): Target host
method (str): HTTP method
uri (str): Request URI
header (dict, optional): Request headers
data (str): Request body data
"""HTTP response formatting utility for logging and debugging.
class ResponsePrettyFormatter:
def __init__(self, resp, format_body: bool = True, delimiter: str = "\n"):
"""
Create response formatter.
Args:
resp: HTTP response object
format_body (bool): Include response body (default: True)
delimiter (str): Line delimiter (default: "\n")
"""
@staticmethod
def str_ver(ver: int) -> str:
"""
Convert HTTP version number to string.
Args:
ver (int): HTTP version number (10, 11, 20)
Returns:
str: HTTP version string ("HTTP/1.0", "HTTP/1.1", "HTTP/2.0")
"""Advanced connection pooling with pre-connected connections for improved performance.
class HTTPSPreConnPool:
def __init__(self, *args, **kwargs):
"""
HTTPS connection pool with pre-connected connections.
Automatically maintains a pool of pre-connected HTTPS connections
for improved performance by reducing connection establishment overhead.
"""
class HTTPPreConnPool:
def __init__(self, *args, **kwargs):
"""
HTTP connection pool with pre-connected connections.
Automatically maintains a pool of pre-connected HTTP connections
for improved performance by reducing connection establishment overhead.
"""
class PreConnPoolManager:
def __init__(self, pool_size: int, *args, **kwargs):
"""
Pool manager for pre-connected connection pools.
Args:
pool_size (int): Maximum number of connections per pool
"""
class PreConnAdapter:
def __init__(self, conn_pool_size: int, *args, **kwargs):
"""
HTTP adapter using pre-connected connection pools.
Args:
conn_pool_size (int): Connection pool size
"""def _get_proxy_from_env(host: str, varname: str = "HTTPS_PROXY") -> str:
"""
Extract proxy configuration from environment variables.
Args:
host (str): Target host to check against NO_PROXY
varname (str): Environment variable name (default: "HTTPS_PROXY")
Returns:
str or None: Proxy URL if found and not excluded, None otherwise
"""from tencentcloud.common.http.request import ApiRequest, RequestInternal
# Create API request handler
request = ApiRequest("api.example.com", req_timeout=30)
# Create request object
req = RequestInternal(
host="api.example.com",
method="POST",
uri="/api/v1/endpoint",
header={"Content-Type": "application/json"},
data='{"key": "value"}'
)
# Send request
try:
response = request.send_request(req)
print(response.status_code)
print(response.text)
except Exception as e:
print(f"Request failed: {e}")from tencentcloud.common.http.request import ApiRequest
# Create client with proxy
request = ApiRequest(
host="api.example.com",
proxy="http://proxy.example.com:8080",
req_timeout=60
)
# Enable keep-alive for better performance
request.set_keep_alive(True)from tencentcloud.common.http.request import ApiRequest
# Create client with pre-connected pool for better performance
request = ApiRequest(
host="api.example.com",
pre_conn_pool_size=10 # Maintain 10 pre-connected connections
)from tencentcloud.common.http.request import ApiRequest
# Use custom certificate bundle
request = ApiRequest(
host="api.example.com",
certification="/path/to/custom/ca-bundle.crt"
)
# Disable certificate verification (not recommended for production)
request = ApiRequest(
host="api.example.com",
certification=False
)from tencentcloud.common.http.request import ApiRequest
# Enable debug logging
request = ApiRequest("api.example.com", debug=True)
# Or enable after creation
request.set_debug(True)import os
from tencentcloud.common.http.request import _get_proxy_from_env
# Set proxy in environment
os.environ["HTTPS_PROXY"] = "http://proxy.example.com:8080"
os.environ["NO_PROXY"] = "localhost,127.0.0.1"
# Get proxy for host (respects NO_PROXY)
proxy = _get_proxy_from_env("api.example.com")
if proxy:
print(f"Using proxy: {proxy}")Install with Tessl CLI
npx tessl i tessl/pypi-tencentcloud-sdk-python-common