Retry-Logik und Verbindungsresilienz mit mssql-python

Vorübergehende Fehler sind temporäre Fehler, die auftreten können, wenn man sich über den mssql-python-Treiber mit SQL Server und Azure SQL verbindet. Diese Fehler lösen sich oft von selbst:

• Kurzzeitige Unterbrechungen der Netzwerkkonnektivität.
• Serverressourcenbeschränkungen.
• Azure SQL-Drosselung.
• Failover-Ereignisse.

Die Implementierung von Retry-Logik verbessert die Zuverlässigkeit der Anwendung, insbesondere bei cloudgehosteten Datenbanken.

Verwende keine Wiederholungen, um Konfigurations- oder Programmierfehler zu verbergen. Eine fehlende Datenbank, schlechte Zugangsdaten oder ein leerer Verbindungspool braucht eine Lösung, keinen weiteren Versuch.

Mit Pooling und dem Standardanbieter msodbcsql18 wendet der Treiber vorübergehend einen nativen Timeout für Verbindungsvorgänge während der Validierung einer wiederverwendeten Verbindung an: den positiven Login-Timeout oder 15 Sekunden, wenn der Login-Timeout null oder nicht spezifiziert ist. Dies ist keine endgültige Validierungs- oder Abschlussfrist. Mehrere native Operationen, Bereinigung und das Öffnen einer Ersatzverbindung können Zeit kosten. Wenn die Validierung fehlschlägt, versucht der Treiber eine Ersetzung, anstatt auf jede verbleibende Leerlaufverbindung zu warten. Diese Wiederherstellung spielt deine Anweisungen nicht erneut ab und versucht die Pool-Erschöpfung nicht erneut. Weitere Informationen finden Sie unter Connection pooling.

Identifizieren Sie transienten Fehler

mssql-python stellt die Fehlernummer der SQL Server-Engine nicht als Attribut bei Ausnahmen offen. Stattdessen ordnet der Treiber SQLSTATE-Codes einer festen Menge von PEP 249-Ausnahmeunterklassen (OperationalError, ProgrammingError, und so weiter) und dem standardisierten englischen Text im Attribut driver_error zu. Verwenden Sie diese Kombination als Grundlage für die vorübergehende Klassifikation.

Zuverlässige transiente Signale

Die folgenden SQLSTATE-Werte werden als OperationalError zurückgegeben und weisen auf einen Zustand hin, bei dem ein erneuter Versuch sinnvoll ist. Die rechte Spalte zeigt den genauen driver_error Text, den der Treiber gesetzt hat:

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-Drosselung (nach bestem Bemühen)

Azure SQL-Throttling-Fehler (40197, 40501, 40613, 49918, 49919, 49920 und verwandte Codes) werden typischerweise mit SQLSTATE 42000 übermittelt, das mssql-python ProgrammingError zuordnet. Die Engine-Fehlernummer wird nicht als Attribut angezeigt, daher ist das einzige Signal der Server-Nachrichtentext im Attribut ddbc_error .

Prüfen Sie ddbc_error, sofern vorhanden, auf bekannte Nummern und auf die englische Standardfehlermeldung 40501. Die Nummer kann fehlen, und der Nachrichtentext ist keine verlässliche Zusicherung. Lokalisierte oder anders formatierte Diagnosedaten könnten nicht übereinstimmen:

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")
_AZURE_40501_MESSAGE = "The service is currently busy. Retry the request after 10 seconds."


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 (
        _AZURE_40501_MESSAGE in ddbc_text
        or 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)

Was nicht wiederholt werden soll

Beispiele für Fehler, die schnell fehlschlagen sollten, anstatt es erneut zu versuchen, sind ungültige Zugangsdaten (OperationalError mit Treibertext Invalid authorization specification), eine fehlende Datenbank oder unzureichende Berechtigungen, Syntaxfehler (ProgrammingError), fehlende Objekte und die Erschöpfung des Verbindungspools. Die obige Funktion is_transient_error schließt all diese durch Konstruktion aus.

Einfacher Wiederholungs-Dekorator

Einfacher Wiederholungsversuch mit fester Verzögerung

Ein Dekorator, der die umschlossene Funktion eine festgelegte Anzahl von Malen mit einer konstanten Verzögerung wiederholt:

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

Exponentiales Backoff

Erhöhen Sie die Verzögerung zwischen den Wiederholungsversuchen exponentiell mit optionalem Jitter, um parallele Wiederholungsversuche zu verteilen:

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

Klasse für Verbindungswiederholungsversuche

Resilienter Verbindungsmanager

Ein Wrapper für Verbindungen, der sowohl Wiederholungsversuche als auch die automatische Wiederverbindung handhabt:

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-spezifische Verarbeitung

Verwenden Sie das mssql-python-Ausnahmemapping in Identify transient errors , um Fehler aus dem folgenden Katalog zu klassifizieren.

Die folgenden Fehler können vorübergehend auftreten, wenn sie während der Verbindungsaufnahme oder beim Senden einer Anfrage an den Server auftreten. Versuchen Sie es mit einem kurzen, begrenzten Backoff. Fehler, die auch nach einigen Wiederholungen bestehen bleiben, deuten meist auf ein Konfigurationsproblem hin, wie einen falschen Server, fehlende Berechtigungen, inkompatible Verschlüsselungseinstellungen oder eine erschöpfte Quote, die ein erneuter Versuch nicht beheben kann.

Fehler auf Anweisungsebene befinden sich nicht in dieser Liste, da sie ausgelöst werden, nachdem die Verbindung hergestellt wurde und der Fehler die Sitzung nutzbar lässt. Die häufigsten Fehler bei Anweisungen, die erneut ausgeführt werden können, sind 1205 (Deadlock-Opfer) und 1222 (Zeitüberschreitung bei der Sperranforderung). Wiederholen Sie die gesamte Transaktion statt der einzelnen fehlerhaften Anweisung.

Fehlermeldungstext stammt aus Azure SQL vorübergehenden Verbindungsfehlern. Diese Fehler sind für Wiederholungen in SQL Server, Azure SQL-Datenbank, Azure SQL Managed Instance, SQL-Datenbank in Microsoft Fabric und dedizierten SQL-Pools in Azure Synapse Analytics zugelassen.

Behandlung von Azure-Drosselung

Azure SQL-Throttling-Fehler benötigen längere Verzögerungen und mehr Wiederholungen als Standard-Transientenfehler. Wiederverwenden is_azure_throttling aus Vorübergehende Fehler identifizieren:

def execute_with_throttle_handling(cursor, query: str, params: dict,
                                   max_retries: int = 10,
                                   base_delay: float = 10.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 16 times the base delay
                    print(f"Throttled. Waiting {delay}s before retry...")
                    time.sleep(delay)
                    continue
            raise

Failover-Behandlung

Verbinden Sie sich erneut und versuchen Sie es erneut, wenn Azure SQL oder Availability Group Failover eine Verbindung unterbricht:

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=AdventureWorks2025;"
    "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"}
)

Behandlung von Deadlocks

Bei Deadlock wiederholen

Deadlocks (Fehler 1205) sind vorübergehend. Versuche es nach einer kurzen zufälligen Verzögerung erneut, um die Deadlock-Schleife zu durchbrechen. Ein erneutes Versuch behebt den sofortigen Fehler, aber wiederkehrende Deadlocks deuten auf ein Designproblem hin, das du serverseitig untersuchen solltest. Für Hinweise zur Analyse und Lösung der Ursache siehe Deadlock-Fehler.

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

Strukturierte Wiederholung mit Konfiguration

Klasse für Wiederholungsrichtlinien

Kapseln Sie die Wiederholungskonfiguration in einer Datenklasse für die Wiederverwendung über verschiedene Operationen hinweg:

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)

Muster „Trennschalter“

Vermeiden Sie kaskadierende Ausfälle, indem Sie aufeinanderfolgende Fehler verfolgen und Aufrufe vorübergehend blockieren, wenn ein Schwellenwert erreicht wird:

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

Konfigurationsfehler nicht erneut versuchen

Nicht jeder Fehler ist vorübergehend. Ein erneuter Versuch einer Konfiguration oder eines Programmierfehlers kostet Zeit und kann das eigentliche Problem kaschieren. Versuchen Sie nur Fehler erneut, die sich von selbst beheben könnten. Da mssql-python die Engine-Fehlernummer nicht als Attribut offenlegt, klassifiziere nach Ausnahme-Unterklasse plus dem driver_error Text.

Versuchen Sie diese nie erneut (korrigieren Sie stattdessen den Code oder die Konfiguration):

Versuche diese immer erneut (sie lösen sich von selbst):

Verwandte Inhalte

• Fehlerbehandlung und SQLSTATE-Codes für mssql-python
• Verbindungspooling mit mssql-python
• Fehlerbehebung mssql-python
• Problembehandlung bei vorübergehenden Verbindungsfehlern