Ponownie spróbuj logiki i odporności połączeń za pomocą mssql-python

Awarie przejściowe to tymczasowe błędy, które mogą wystąpić podczas łączenia się z SQL Server i Azure SQL za pomocą sterownika mssql-python. Te błędy często rozwiązują się same:

• Krótkotrwałe zakłócenia łączności sieciowej.
• Ograniczenia zasobów serwera.
• Azure SQL throttling.
• Zdarzenia awaryjne.

Wdrożenie logiki powtórek poprawia niezawodność aplikacji, zwłaszcza w bazach danych hostowanych w chmurze.

Nie używaj powtórek do ukrywania błędów konfiguracyjnych lub kodowych. Brakująca baza danych, złe dane uwierzytelniające lub wyczerpana pula połączeń wymaga naprawy, a nie kolejnej próby.

Identyfikacja błędów przejściowych

mssql-python nie ujawnia numeru błędu silnika SQL Server jako atrybutu w przypadku wyjątków. Zamiast tego sterownik mapuje kody SQLSTATE na stały zestaw podklas wyjątków PEP 249 (OperationalError, ProgrammingError, i tak dalej) oraz na ustandaryzowany angielski tekst w atrybucie driver_error . Użyj tej kombinacji jako podstawy klasyfikacji przejściowych.

Niezawodne sygnały przejściowe

Następujące wartości SQLSTATE pojawiają się jako OperationalError i wskazują warunek warujący ponowne wypróbowanie. Prawa kolumna pokazuje dokładny driver_error tekst ustawiony przez sterownik:

SQLSTATE	driver_error Tekst	Warunek
HYT00	Timeout expired	Limit czasu na poziomie instrukcji.
HYT01	Connection timeout expired	Limit czasu połączenia.
08001	Client unable to establish connection	Nie udało się nawiązać połączenia.
08S01	Communication link failure	Zerwanie sieci, reset serwera, awaria TCP.
08007	Connection failure during transaction	Połączenie zerwane w trakcie transakcji.
40001	Serialization failure	Ofiara zakleszczenia.
40003	Statement completion unknown	Stan transakcji nieokreślony.

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

Ograniczanie przepustowości usługi Azure SQL (w miarę możliwości)

Błędy ograniczania przepustowości w Azure SQL (40197, 40501, 40613, 49918, 49919, 49920 i powiązane kody) są zazwyczaj zgłaszane z kodem SQLSTATE 42000, który mssql-python odwzorowuje jako ProgrammingError. Numer błędu silnika nie jest wyświetlany jako atrybut, więc jedynym sygnałem jest tekst wiadomości serwera w atrybucie ddbc_error .

Jeśli Twoje obciążenie korzysta z usługi Azure SQL i musisz ponawiać próby po ograniczeniu przepustowości, przeszukaj ddbc_error, aby znaleźć znany numer. To jest najlepszy wysiłek, ponieważ format tekstu po stronie serwera nie jest stabilnym kontraktem:

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)

Czego nie ponawiać

Przykłady błędów, które powinny od razu kończyć się niepowodzeniem zamiast ponawiania prób, obejmują nieprawidłowe poświadczenia uwierzytelniające (OperationalError z tekstem sterownika Invalid authorization specification), brakującą lub niedostępną bazę danych, błędy składniowe (ProgrammingError), brakujące obiekty oraz wyczerpanie puli połączeń. Powyższa is_transient_error funkcja wyklucza wszystkie te elementy z konstrukcji.

Podstawowy dekorator ponawiania

Prosta próba z stałym opóźnieniem

Dekorator, który ponawia wywołanie dekorowanej funkcji określoną liczbę razy ze stałym opóźnieniem:

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

Wycofywanie wykładnicze

Zwiększ opóźnienie między próbami wykładniczo za pomocą opcjonalnego jittera, aby rozłożyć równoczesne próby:

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

Klasa powtórki połączeń

Menedżer trwałych połączeń

Opakowanie połączenia, które obsługuje zarówno próbę ponowną, jak i automatyczne ponowne połączenie:

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

Obsługa specyficzna dla Azure SQL

Obsługa ograniczania przepustowości w usłudze Azure

Błędy ograniczania przepustowości w usłudze Azure SQL wymagają dłuższych opóźnień i większej liczby ponownych prób niż typowe błędy przejściowe. Użyj ponownie is_azure_throttling z Identyfikowanie błędów przejściowych:

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

Obsługa trybu failover

Połącz się ponownie i spróbuj ponownie, gdy przełączenie awaryjne usługi Azure SQL lub grupy dostępności przerwie połączenie:

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

Obsługa zakleszczeń

Ponów próbę w przypadku zakleszczenia

Blokady (błąd 1205) są przejściowe. Spróbuj ponownie po krótkim losowym opóźnieniu, aby przerwać cykl zakleszczenia. Ponowne próbowanie rozwiązuje natychmiastową awarię, ale powtarzające się zablokowania wskazują na problem projektowy, który warto zbadać po stronie serwera. Aby uzyskać wskazówki dotyczące analizy i rozwiązania podstawowej przyczyny, zobacz błędy zakleszczenia.

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

Strukturalne ponowne próbowanie z konfiguracją

Klasa zasad ponawiania

Enkapsuluj konfigurację prób powtórnych w klasie danych do ponownego wykorzystania w różnych operacjach:

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)

Wzorzec wyłącznika

Zapobieganie kaskadowym awariom poprzez śledzenie kolejnych błędów i tymczasowe blokowanie wywołań po osiągnięciu progu:

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

Treści powiązane

• Obsługa błędów
• Buforowanie połączeń
• Troubleshooting
• Odporność połączenia Azure SQL

Nie próbuj ponownie błędów konfiguracji

Nie każdy błąd jest przejściowy. Ponowne próbowanie błędu konfiguracji lub kodu marnuje czas i może ukryć prawdziwy problem. Powtarzaj tylko błędy, które mogą same się rozwiązać. Ponieważ mssql-python nie udostępnia numeru błędu silnika jako atrybutu, klasyfikuj na podstawie podklasy wyjątku oraz tekstu driver_error.

Nigdy nie próbuj ponownie ( napraw kod lub konfigurację):

Warunek	Typ wyjątku	driver_error Tekst	Napraw.
Nieprawidłowa nazwa obiektu (silnik 208)	ProgrammingError	Base table or view not found	Tabela nie istnieje. Napraw zapytanie lub stwórz tabelę.
Nieprawidłowa nazwa kolumny (silnik 207)	ProgrammingError	Column not found	Kolumna nie istnieje. Sprawdź schemat.
Nieprawidłowa składnia (engine 102)	ProgrammingError	Syntax error or access violation	Napraw zapytanie.
Logowanie nieudane (silnik 18456)	OperationalError	Invalid authorization specification	Błędne kwalifikacje. Napraw parametry połączenia.
Nie można otworzyć bazy danych (engine 4060)	OperationalError	Server rejected the connection	Baza danych nie istnieje lub nie jest dostępna przez logowanie. Napraw cel lub uprawnienia.
Wyczerpanie puli połączeń	OperationalError	(różni się)	Zwiększ pojemność puli, szybko zwolnij połączenia lub ogranicz współbieżność.
ConnectionStringParseError	Samodzielny	nie dotyczy	Literówka w słowie kluczowym parametry połączenia. Napraw sznurek.
Funkcja nieobsługiwana	NotSupportedError	Optional feature not implemented	Stosuj alternatywne podejście.

Zawsze próbuj ponownie ( same się rozwiązują):

Warunek	Typ wyjątku	driver_error Tekst
Limit czasu instrukcji	OperationalError	Timeout expired
Limit czasu połączenia	OperationalError	Connection timeout expired
Nie udało się nawiązać połączenia	OperationalError	Client unable to establish connection
Przerwanie sieci	OperationalError	Communication link failure
Połączenie zerwane w trakcie transakcji	OperationalError	Connection failure during transaction
Ofiara zakleszczenia (silnik 1205)	OperationalError	Serialization failure
Stan transakcji nieokreślony	OperationalError	Statement completion unknown
Azure SQL throttling (40197, 40501, 40613, 49918–49920)	ProgrammingError	Syntax error or access violation (numer silnika jest tylko w ddbc_error; użyj is_azure_throttling)