Probeer opnieuw logica en verbindingsveerkracht met mssql-python

Tijdelijke storingen zijn tijdelijke fouten die kunnen optreden bij het verbinden met SQL Server en Azure SQL via de mssql-python-driver. Deze fouten lossen vaak vanzelf op:

• Korte netwerkonderbrekingen.
• Serverresourcebeperkingen.
• Azure SQL throttling.
• Failover-gebeurtenissen.

Het implementeren van hertelingslogica verbetert de betrouwbaarheid van applicaties, vooral voor cloud-gehoste databases.

Gebruik geen nieuwe pogingen om configuratie- of programmeerfouten te verbergen. Een ontbrekende database, slechte inloggegevens of een uitgeputte verbindingspool heeft een oplossing nodig, geen nieuwe poging.

Bij pooling en de standaardprovider msodbcsql18 past de driver tijdelijk een native timeout voor verbindingsoperaties toe tijdens de validatie van een hergebruikte verbinding: de positieve logintimeout, oftewel 15 seconden wanneer de logintimeout nul of niet gespecificeerd is. Dit is geen volledige validatie of deadline voor afrekenen. Meerdere native bewerkingen, opruimen en het openen van een vervangende verbinding kunnen extra tijd kosten. Als validatie faalt, probeert de driver een vervangende verbinding in plaats van te wachten op elke resterende idle verbinding. Dit herstel speelt je statements niet opnieuw af en herhaalt uitputting van de retrypool niet. Voor meer informatie, zie Connection pooling.

Identificeer tijdelijke fouten

mssql-python geeft het foutnummer van de SQL Server-engine niet als attribuut op uitzonderingen. In plaats daarvan koppelt de driver SQLSTATE-codes aan een vaste set PEP 249 uitzonderingssubklassen (OperationalError, ProgrammingError, enzovoort) en aan gestandaardiseerde Engelse tekst in het driver_error attribuut. Gebruik die combinatie als basis voor tijdelijke classificatie.

Betrouwbare transiënte signalen

De volgende SQLSTATE-waarden worden doorgegeven als OperationalError en duiden op een toestand waarbij het zinvol is om het opnieuw te proberen. De rechterkolom toont de exacte driver_error tekst die door de driver is ingesteld:

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-beperking (naar beste vermogen)

Azure SQL-throttlingfouten (40197, 40501, , 40613, 49918, 49919, , 49920, en gerelateerde codes) komen meestal voor bij SQLSTATE 42000, dat mssql-python koppelt aan ProgrammingError. Het enginefoutnummer wordt niet als attribuut weergegeven, dus het enige signaal is de serverberichttekst in het ddbc_error attribuut.

Controleer ddbc_error op bekende nummers wanneer aanwezig en op het standaard Engelse foutbericht 40501. Het nummer kan ontbreken, en de berichttekst is geen stabiel contract. Gelokaliseerde of anders geformatteerde diagnostiek komt mogelijk niet overeen:

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)

Wat u niet opnieuw moet proberen

Voorbeelden van fouten die direct moeten mislukken in plaats van opnieuw te worden geprobeerd, zijn onder andere ongeldige inloggegevens (OperationalError met drivertekst Invalid authorization specification), een ontbrekende database of onvoldoende permissies, syntaxisfouten (ProgrammingError), ontbrekende objecten en uitputting van de verbindingspool. De is_transient_error bovenstaande functie sluit al deze uit door constructie.

Basale decorator voor opnieuw proberen

Eenvoudige herpoging met vaste vertraging

Een decorator die de ingepakte functie een vast aantal keer opnieuw uitvoert met een constante vertraging:

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

Exponentiële terugschakeling

Verhoog de vertraging tussen herhalingen exponentieel met optionele jitter om gelijktijdige herhalingen te spreiden:

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 voor opnieuw verbinden

Veerkrachtige verbindingsmanager

Een verbindingswrapper die zowel herpoging als automatische herverbinding afhandelt:

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-specifieke afhandeling

Gebruik de mssql-python exception mapping in Identify transient errors om fouten uit de volgende catalogus te classificeren.

De volgende fouten kunnen voorbijgaand zijn wanneer ze optreden tijdens het tot stand brengen van de verbinding of tijdens het verzenden van een verzoek naar de server. Probeer het na een korte, begrensde wachttijd opnieuw. Fouten die langer dan een paar herpogingen blijven bestaan, wijzen meestal op een configuratieprobleem, zoals een verkeerde server, ontbrekende permissies, incompatibele encryptie-instellingen of een uitgeputte quotum, die een herpoging niet oplost.

Fouten op instructieniveau staan niet in deze lijst, omdat ze optreden nadat de verbinding tot stand is gekomen en de sessie na de fout bruikbaar blijft. De meest voorkomende fouten in opnieuw uit te voeren instructies zijn 1205 (deadlockslachtoffer) en 1222 (time-out bij een vergrendelingsaanvraag). Voer de hele transactie opnieuw uit in plaats van de enkele mislukte instructie.

De tekst van het foutbericht is afkomstig van Azure SQL tijdelijke verbindingsfouten. Deze fouten zijn geschikt voor herhaling via SQL Server, Azure SQL Database, Azure SQL Managed Instance, SQL database in Microsoft Fabric en speciale SQL-pools in Azure Synapse Analytics.

Azure-throttling afhandelen

Azure SQL throttling-fouten vereisen langere vertragingen en meer herpogingen dan standaard transient-fouten. Hergebruik is_azure_throttling van Herken tijdelijke fouten:

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 afhandelen

Maak opnieuw verbinding en probeer opnieuw wanneer Azure SQL of availability group failover een verbinding onderbreekt:

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"}
)

Afhandeling van deadlocks

Opnieuw proberen bij patstelling

Deadlocks (fout 1205) zijn tijdelijk. Probeer opnieuw met een korte willekeurige vertraging om de deadlock-cyclus te doorbreken. Opnieuw proberen lost de directe storing op, maar terugkerende deadlocks wijzen op een ontwerpprobleem dat je serverzijde moet onderzoeken. Voor richtlijnen bij het analyseren en oplossen van de onderliggende oorzaak, zie Deadlock-fouten.

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

Gestructureerde herpoging met configuratie

Retry policy-klasse

Encapsuleer de configuratie van retry in een dataclass voor hergebruik over verschillende bewerkingen:

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)

Circuitonderbrekerpatroon

Voorkom cascaderende storingen door opeenvolgende fouten te volgen en tijdelijk aanroepen te blokkeren wanneer een drempel wordt bereikt:

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

Probeer configuratiefouten niet opnieuw

Niet elke fout is voorbijgaand. Het opnieuw proberen van een configuratie- of codefout verspilt tijd en kan het echte probleem verhullen. Probeer alleen fouten opnieuw die vanzelf kunnen worden opgelost. Omdat mssql-python het foutnummer van de engine niet als attribuut blootlegt, classificeer je op uitzonderingssubklasse plus de driver_error tekst.

Probeer deze nooit opnieuw (repareer de code of configuratie):

Probeer deze altijd opnieuw (ze lossen vanzelf op):

Verwante onderwerpen

• Foutafhandeling en SQLSTATE-codes voor mssql-python
• Verbindingspooling met mssql-python
• Problemen oplossen met mssql-python
• Tijdelijke verbindingsfouten oplossen