用 MSSQL-Python 重试逻辑和连接弹性

暂时性故障是指通过mssql-python驱动连接SQL Server和Azure SQL时可能发生的临时错误。 这些错误通常会自行解决：

• 网络连接短暂中断。
• 服务器资源限制。
• Azure SQL throttling.
• 故障切换事件。

实现重试逻辑可以提升应用的可靠性，尤其是对于云托管数据库。

不要用重试来隐藏配置或编码错误。 数据库缺失、凭证错误或连接池耗尽需要修复，而不是再尝试一次。

识别瞬态错误

mssql-python 不会在异常时将 SQL Server 引擎的错误编号作为属性暴露。 相反，驱动程序将 SQLSTATE 代码映射到一组固定的 PEP 249 例外子类（OperationalError、、 ProgrammingError、等）以及属性中的 driver_error 标准化英文文本。 将该组合作为瞬态分类的基础。

可靠的瞬态信号

以下 SQLSTATE 值会以 OperationalError 形式返回，表示某种值得重试的情况。 右侧列显示了司机设置的具体 driver_error 文字：

import mssql_python


TRANSIENT_DRIVER_ERRORS = frozenset({
    "Timeout expired",
    "Connection timeout expired",
    "Client unable to establish connection",
    "Communication link failure",
    "Connection failure during transaction",
    "Serialization failure",
    "Statement completion unknown",
})


def is_transient_error(error: BaseException) -> bool:
    """Return True if the exception represents a retryable transient failure.

    Classification is based on the driver's PEP 249 exception subclass and
    on the standardized `driver_error` text that mssql-python sets from
    the SQLSTATE returned by the server.
    """
    if isinstance(error, mssql_python.OperationalError):
        return getattr(error, "driver_error", "") in TRANSIENT_DRIVER_ERRORS
    return False

Azure SQL 限制（尽力而为）

Azure SQL 限流错误（40197、40501、40613、49918、49919、49920及相关代码）通常会以 SQLSTATE 42000 的形式出现，mssql-python 会将其映射为 ProgrammingError。 引擎错误编号不会作为属性显示，所以唯一的信号是属性中的 ddbc_error 服务器消息文本。

如果你的工作负载在 Azure SQL 上运行，并且在遇到节流时需要重试，请查看 ddbc_error 中的已知编号。 这是尽力而为，因为服务器端文本的格式不是稳定的契约：

import re

# Azure SQL throttling and reconfiguration error numbers.
AZURE_THROTTLING_ERRORS = frozenset({
    40197, 40501, 40540, 40613, 40680, 49918, 49919, 49920, 10928, 10929,
})

_ERROR_NUMBER_RE = re.compile(r"\b(?:Error|Msg)\s+(\d+)\b")


def is_azure_throttling(error: BaseException) -> bool:
    """Best-effort detection of Azure SQL throttling in ProgrammingError text."""
    if not isinstance(error, mssql_python.ProgrammingError):
        return False
    ddbc_text = getattr(error, "ddbc_error", "") or ""
    return any(int(m) in AZURE_THROTTLING_ERRORS for m in _ERROR_NUMBER_RE.findall(ddbc_text))


def is_retryable(error: BaseException) -> bool:
    return is_transient_error(error) or is_azure_throttling(error)

哪些内容不应重试

应快速失败而非重试的错误示例包括无效凭证（OperationalError 带驱动文本 Invalid authorization specification）、缺失或无法访问的数据库、语法错误（）、ProgrammingError缺少对象以及连接池耗尽。 is_transient_error上述函数通过构造排除了所有这些。

基本重试装饰器

简单固定延迟重试

一种装饰器，它会以固定延迟对被包装的函数重试固定次数：

import time
import functools
import mssql_python

def retry_on_failure(max_retries: int = 3, delay: float = 1.0):
    """Decorator to retry database operations on transient failures."""
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            last_exception = None
            for attempt in range(max_retries + 1):
                try:
                    return func(*args, **kwargs)
                except mssql_python.Error as e:
                    last_exception = e
                    if not is_transient_error(e) or attempt == max_retries:
                        raise
                    print(f"Attempt {attempt + 1} failed: {e}. Retrying in {delay}s...")
                    time.sleep(delay)
            raise last_exception
        return wrapper
    return decorator

# Usage
@retry_on_failure(max_retries=3, delay=2.0)
def get_user(cursor, user_id: int):
    cursor.execute("SELECT * FROM Person.Person WHERE BusinessEntityID = %(id)s", {"id": user_id})
    return cursor.fetchone()

指数退避

通过可选抖动将重试间隔呈指数增加，以分散并发重试时间：

import time
import random

def retry_with_backoff(max_retries: int = 5, 
                       base_delay: float = 1.0,
                       max_delay: float = 30.0,
                       jitter: bool = True):
    """Retry with exponential backoff and optional jitter."""
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            last_exception = None
            for attempt in range(max_retries + 1):
                try:
                    return func(*args, **kwargs)
                except mssql_python.Error as e:
                    last_exception = e
                    if not is_transient_error(e) or attempt == max_retries:
                        raise
                    
                    # Calculate delay with exponential backoff
                    delay = min(base_delay * (2 ** attempt), max_delay)
                    if jitter:
                        delay = delay * (0.5 + random.random())
                    
                    print(f"Attempt {attempt + 1} failed. Retrying in {delay:.2f}s...")
                    time.sleep(delay)
            raise last_exception
        return wrapper
    return decorator

@retry_with_backoff(max_retries=5, base_delay=1.0, max_delay=30.0)
def execute_query(cursor, query: str, params: dict):
    cursor.execute(query, params)
    return cursor.fetchall()

连接重试类

弹性连接管理器

一个既能处理重试也处理自动重连的连接封装器：

import mssql_python
import time
import logging

# This example uses is_transient_error from the "Identify transient errors"
# section earlier in this article. Include that helper in your module.

# Configure logging so the retry and reconnect messages are visible
logging.basicConfig(level=logging.INFO)

class ResilientConnection:
    """Connection wrapper with automatic retry and reconnection."""
    
    def __init__(self, connection_string: str, max_retries: int = 5,
                 base_delay: float = 1.0, max_delay: float = 60.0):
        self.connection_string = connection_string
        self.max_retries = max_retries
        self.base_delay = base_delay
        self.max_delay = max_delay
        self._conn = None
        self._logger = logging.getLogger(__name__)
    
    def _connect(self) -> mssql_python.Connection:
        """Establish connection with retry logic."""
        last_exception = None
        
        for attempt in range(self.max_retries + 1):
            try:
                self._logger.debug(f"Connection attempt {attempt + 1}")
                return mssql_python.connect(self.connection_string)
            except mssql_python.Error as e:
                last_exception = e
                if not is_transient_error(e) or attempt == self.max_retries:
                    self._logger.error(f"Connection failed: {e}")
                    raise
                
                delay = min(self.base_delay * (2 ** attempt), self.max_delay)
                self._logger.warning(f"Connection attempt {attempt + 1} failed. "
                                   f"Retrying in {delay:.1f}s...")
                time.sleep(delay)
        
        raise last_exception
    
    @property
    def connection(self) -> mssql_python.Connection:
        """Get or create connection."""
        if self._conn is None:
            self._conn = self._connect()
        return self._conn
    
    def execute(self, query: str, params: dict = None):
        """Execute query with automatic retry and reconnection."""
        return self._execute_with_retry(
            lambda c: self._do_execute(c, query, params)
        )
    
    def _do_execute(self, cursor, query: str, params: dict):
        cursor.execute(query, params or {})
        return cursor.fetchall()
    
    def _execute_with_retry(self, operation):
        """Execute an operation with retry logic."""
        last_exception = None
        
        for attempt in range(self.max_retries + 1):
            try:
                cursor = self.connection.cursor()
                return operation(cursor)
            except mssql_python.Error as e:
                last_exception = e
                
                if not is_transient_error(e):
                    raise
                
                if attempt == self.max_retries:
                    raise
                
                # Try to reconnect
                self._logger.warning(f"Operation failed. Reconnecting...")
                self._close()
                
                delay = min(self.base_delay * (2 ** attempt), self.max_delay)
                time.sleep(delay)
        
        raise last_exception
    
    def _close(self):
        """Close connection."""
        if self._conn:
            try:
                self._conn.close()
            except:
                pass
            self._conn = None
    
    def close(self):
        """Public close method."""
        self._close()
    
    def __enter__(self):
        return self
    
    def __exit__(self, exc_type, exc_val, exc_tb):
        self.close()
        return False

# Usage
with ResilientConnection(connection_string) as db:
    users = db.execute("SELECT * FROM Person.Person WHERE EmailPromotion = %(promo)s", 
                       {"promo": 1})
    print(f"Retrieved {len(users)} rows")

Azure SQL 特定处理

处理 Azure 限流

Azure SQL 限流错误需要比标准瞬态错误更长的延迟和更多的重试。 从is_azure_throttling中重用：

def execute_with_throttle_handling(cursor, query: str, params: dict,
                                   max_retries: int = 10,
                                   base_delay: float = 5.0):
    """Execute with extended retry for Azure SQL throttling."""
    for attempt in range(max_retries + 1):
        try:
            cursor.execute(query, params)
            return cursor.fetchall()
        except mssql_python.Error as e:
            if is_azure_throttling(e):
                if attempt < max_retries:
                    # Longer delays for throttling
                    delay = base_delay * (2 ** min(attempt, 4))  # Cap at 80s
                    print(f"Throttled. Waiting {delay}s before retry...")
                    time.sleep(delay)
                    continue
            raise

处理故障转移

当 Azure SQL 或可用性组故障切换导致连接中断时，请重新连接并重试：

def execute_with_failover_retry(connect, query: str, params: dict,
                                max_retries: int = 3,
                                recovery_delay: float = 10.0):
    """Reconnect and retry during Azure SQL failover scenarios."""
    failover_numbers = frozenset({40613, 40197, 40540})
    last_exception = None

    for attempt in range(max_retries + 1):
        conn = None
        try:
            conn = connect()
            cursor = conn.cursor()
            cursor.execute(query, params)
            return cursor.fetchall()
        except mssql_python.Error as e:
            last_exception = e

            # Failover surfaces either as a transient OperationalError or as
            # a ProgrammingError whose ddbc_error text contains the engine
            # error number. Treat both as recoverable.
            ddbc_text = getattr(e, "ddbc_error", "") or ""
            is_failover = is_transient_error(e) or any(
                int(m) in failover_numbers for m in _ERROR_NUMBER_RE.findall(ddbc_text)
            )

            if is_failover and attempt < max_retries:
                print(f"Failover detected. Reconnecting in {recovery_delay}s...")
                if conn is not None:
                    try:
                        conn.close()
                    except mssql_python.Error:
                        pass
                time.sleep(recovery_delay)
                continue
            raise

    raise last_exception


# Usage
connection_string = (
    "Server=tcp:<server>.database.windows.net,1433;"
    "Database=AdventureWorks2022;"
    "Authentication=ActiveDirectoryDefault;"
    "Encrypt=yes;TrustServerCertificate=no"
)

rows = execute_with_failover_retry(
    lambda: mssql_python.connect(connection_string),
    "SELECT TOP 10 ProductID, Name FROM Production.Product WHERE Color = %(color)s",
    {"color": "Silver"}
)

死锁处理

发生死锁时重试

死锁（错误1205）是暂时性的。 用一个短暂的随机延迟重试以打破僵局循环。 重试可以解决立即失败的问题，但反复出现的死锁说明设计存在问题，需要在服务器端调查。 有关分析和解决根本原因的指导，请参见 “死锁错误”。

def execute_with_deadlock_retry(cursor, query: str, params: dict,
                                max_retries: int = 3):
    """Automatically retry deadlocked transactions.

    Deadlocks (SQL Server error 1205) surface as OperationalError with
    driver_error == "Serialization failure" (SQLSTATE 40001).
    """
    for attempt in range(max_retries + 1):
        try:
            cursor.execute(query, params)
            return cursor.fetchall()
        except mssql_python.OperationalError as e:
            if getattr(e, "driver_error", "") == "Serialization failure":
                if attempt < max_retries:
                    delay = random.uniform(0.1, 0.5) * (attempt + 1)
                    print(f"Deadlock detected. Retry {attempt + 1} in {delay:.2f}s")
                    time.sleep(delay)
                    continue
            raise

# Usage in transaction
conn.autocommit = False
try:
    cursor = conn.cursor()
    rows = execute_with_deadlock_retry(
        cursor,
        "SELECT TOP 5 Name, ListPrice FROM Production.Product WHERE ListPrice > %(price)s",
        {"price": 100}
    )
    conn.commit()
except Exception:
    conn.rollback()
    raise

结构化重试与配置

重试策略类

将重试配置封装在数据类中，以便在不同操作间重复使用：

from dataclasses import dataclass, field
from typing import FrozenSet
import time
import random

# This example uses TRANSIENT_DRIVER_ERRORS from the "Identify transient errors"
# section earlier in this article. Include that allowlist in your module.

@dataclass
class RetryPolicy:
    """Configuration for retry behavior."""
    max_retries: int = 3
    base_delay: float = 1.0
    max_delay: float = 30.0
    exponential_base: float = 2.0
    jitter: bool = True
    transient_driver_errors: FrozenSet[str] = field(default_factory=lambda: TRANSIENT_DRIVER_ERRORS)

    def get_delay(self, attempt: int) -> float:
        """Calculate delay for given attempt number."""
        delay = min(
            self.base_delay * (self.exponential_base ** attempt),
            self.max_delay,
        )
        if self.jitter:
            delay *= (0.5 + random.random())
        return delay

    def should_retry(self, error: BaseException, attempt: int) -> bool:
        """Determine if operation should be retried."""
        if attempt >= self.max_retries:
            return False
        if isinstance(error, mssql_python.OperationalError):
            return getattr(error, "driver_error", "") in self.transient_driver_errors
        return False

def execute_with_policy(cursor, query: str, params: dict,
                        policy: RetryPolicy = None):
    """Execute query with configurable retry policy."""
    policy = policy or RetryPolicy()
    last_exception = None

    for attempt in range(policy.max_retries + 1):
        try:
            cursor.execute(query, params)
            return cursor.fetchall()
        except mssql_python.Error as e:
            last_exception = e
            if not policy.should_retry(e, attempt):
                raise

            delay = policy.get_delay(attempt)
            time.sleep(delay)

    raise last_exception

# Usage with custom policy
aggressive_retry = RetryPolicy(max_retries=10, base_delay=0.5, max_delay=60.0)
conservative_retry = RetryPolicy(max_retries=2, base_delay=5.0, max_delay=10.0)

results = execute_with_policy(cursor, query, params, aggressive_retry)

断路器设计模式

通过跟踪连续错误并在达到阈值时暂时阻断呼叫，防止连锁故障：

import time
from enum import Enum
from threading import Lock

class CircuitState(Enum):
    CLOSED = "closed"      # Normal operation
    OPEN = "open"          # Failing, reject all calls
    HALF_OPEN = "half_open"  # Testing if service recovered

class CircuitBreaker:
    """Circuit breaker to prevent cascading failures."""
    
    def __init__(self, failure_threshold: int = 5,
                 recovery_timeout: float = 30.0):
        self.failure_threshold = failure_threshold
        self.recovery_timeout = recovery_timeout
        self.state = CircuitState.CLOSED
        self.failure_count = 0
        self.last_failure_time = None
        self._lock = Lock()
    
    def can_execute(self) -> bool:
        """Check if circuit allows execution."""
        with self._lock:
            if self.state == CircuitState.CLOSED:
                return True
            
            if self.state == CircuitState.OPEN:
                # Check if recovery timeout has passed
                if time.time() - self.last_failure_time > self.recovery_timeout:
                    self.state = CircuitState.HALF_OPEN
                    return True
                return False
            
            # HALF_OPEN: allow one test request
            return True
    
    def record_success(self):
        """Record successful operation."""
        with self._lock:
            self.failure_count = 0
            self.state = CircuitState.CLOSED
    
    def record_failure(self):
        """Record failed operation."""
        with self._lock:
            self.failure_count += 1
            self.last_failure_time = time.time()
            
            if self.failure_count >= self.failure_threshold:
                self.state = CircuitState.OPEN

# Usage
circuit = CircuitBreaker(failure_threshold=5, recovery_timeout=30.0)

def execute_with_circuit_breaker(cursor, query: str, params: dict):
    if not circuit.can_execute():
        raise Exception("Circuit breaker is open")
    
    try:
        cursor.execute(query, params)
        result = cursor.fetchall()
        circuit.record_success()
        return result
    except mssql_python.Error as e:
        if is_transient_error(e):
            circuit.record_failure()
        raise

相关内容

• 错误处理
• 连接池
• Troubleshooting
• Azure SQL 连接复原能力

出现配置错误时不要重试

并非所有错误都是暂时的。 重试配置或编码错误浪费时间，可能掩盖真正的问题。 只重试那些可能自行解决的错误。 因为 mssql-python 不把引擎错误编号作为属性暴露，所以可以按异常子类加上 driver_error 文本分类。

切勿重试这些（ 改为修正代码或配置）：

一定要重试这些（ 它们会自行解决）：