mssql-python ile mantık ve bağlantı dayanıklılığını yeniden deneyin

Geçici hatalar, mssql-python sürücüsü üzerinden SQL Server ve Azure SQL'e bağlanırken meydana gelen geçici hatalardır. Bu hatalar genellikle kendi kendine çözülür:

• Ağ bağlantısında kısa süreli kesintiler.
• Sunucu kaynak kısıtlamaları.
• Azure SQL kısıtlaması
• Failover olayları.

Yeniden deneme mantığının uygulanması, özellikle bulut barındırılan veritabanları için uygulama güvenilirliğini artırır.

Yapılandırma veya kodlama hatalarını gizlemek için tekrar denemeler yapmayın. Eksik bir veritabanı, kötü kimlik bilgileri veya tükenmiş bağlantı havuzu bir düzeltme gerektirir, başka bir deneme değil.

Geçici hataları tespit edin

mssql-python, SQL Server motoru hata numarasını istisnalarda bir öznitelik olarak göstermez. Bunun yerine, sürücü SQLSTATE kodlarını sabit bir PEP 249 istisna alt sınıfları kümesine (OperationalError, ProgrammingError, ve benzeri) ve öznitelikteki driver_error standartlaştırılmış İngilizce metnine eşler. Bu kombinasyonu geçici sınıflandırmanın temeli olarak kullanın.

Güvenilir geçici sinyaller

Aşağıdaki SQLSTATE değerleri OperationalError olarak döner ve yeniden denenmeye değer bir durumu gösterir. Sağ sütun, sürücünün belirlediği tam driver_error metni gösterir:

SQLSTATE	driver_error metin	Koşul
HYT00	Timeout expired	Açıklama seviyesinde mola verme.
HYT01	Connection timeout expired	Bağlantı zamanı molası.
08001	Client unable to establish connection	Bağlantı açamadım.
08S01	Communication link failure	Ağ kesilmesi, sunucu sıfırlanması, TCP arızası.
08007	Connection failure during transaction	İşlem sırasında bağlantı kayboldu.
40001	Serialization failure	Kilitlenme kurbanı.
40003	Statement completion unknown	İşlem durumu belirsiz.

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 kısıtlaması (en iyi çaba düzeyinde)

Azure SQL kısıtlama hataları (40197, 40501, 40613, 49918, 49919, 49920 ve ilgili kodlar) genellikle SQLSTATE 42000 ile gelir ve mssql-python bunu ProgrammingError olarak eşler. Motor hata numarası bir öznitelik olarak ortaya çıkmaz, bu yüzden tek sinyal öznitelikteki ddbc_error sunucu mesaj metnidir.

Eğer iş yükünüz Azure SQL'e karşı çalışıyorsa ve throttling'i tekrar denemeniz gerekiyorsa, ddbc_error bilinen sayı için tara edin. Bu en iyi çabadır çünkü sunucu tarafı metnin formatı kararlı bir sözleşme değildir:

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)

Yeniden denenmemesi gerekenler

Yeniden denemek yerine hızlıca başarısız olması gereken hata örnekleri arasında geçersiz kimlik bilgileri (OperationalError sürücü metni Invalid authorization specificationile), eksik veya erişilemeyen veritabanı, sözdizimi hataları (ProgrammingError), eksik nesneler ve bağlantı havuzunun tükenmesi yer alır. Yukarıdaki fonksiyon, is_transient_error bunların hepsini yapı gereği dışlar.

Temel yeniden deneme dekoratörü

Sabit gecikmeli basit yeniden deneme

Sarmalanan fonksiyonu sabit bir gecikmeyle belirli sayıda yeniden deneyen bir dekoratör:

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

Üstel geri çekilme

İsteğe bağlı jitter kullanarak, aynı anda yapılan yeniden denemeleri zamana yaymak için yeniden denemeler arasındaki gecikmeyi üstel olarak artırın:

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

Bağlantı yeniden deneme sınıfı

Dayanıklı bağlantı yöneticisi

Hem yeniden deneme hem de otomatik yeniden bağlantı işlemlerini yöneten bir bağlantı wrapperı:

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'ye özgü işlem

Azure kısıtlamasını işleme

Azure SQL throttling hataları, standart geçici hatalara göre daha uzun gecikmeler ve daha fazla tekrar deneme gerektirir. is_azure_throttling öğesini Geçici hataları belirleme bölümünden yeniden kullanın:

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

Yük devretmeyi yönetme

Azure SQL veya availability group failover bağlantıyı kestiğinde yeniden bağlan ve tekrar dene:

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

Kilitlenme işleme

Kilitlenme durumunda yeniden dene

Kilitlenmeler (hata 1205) geçicidir. Deadlock döngüsünü kırmak için kısa bir rastgele gecikmeyle tekrar dene. Tekrar denemek anında arızayı yönetir, ancak tekrarlayan çıkmazlıklar sunucu tarafında araştırmanız gereken bir tasarım sorununu gösterir. Kök nedeni analiz etme ve giderme konusunda bilgi için bkz. Deadlock hataları.

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

Konfigürasyonla yapılandırılmış yeniden deneme

Politika sınıfını yeniden deneme

Farklı işlemler arasında yeniden kullanılmak üzere bir veri sınıfında yeniden deneme yapılandırmasını kapsülle:

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)

Devre kesici deseni

Ardışık hataları takip ederek ve bir eşik noktasına ulaşıldığında çağrıları geçici olarak engelleyerek zincirleme hataları önleyin:

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

İlgili içerik

• Hata işleme
• Bağlantı havuzu oluşturma
• Troubleshooting
• Azure SQL bağlantı dayanıklılığı

Yapılandırma hatalarını tekrar denemeyin

Her hata geçici değildir. Bir yapılandırma veya kodlama hatasını tekrar denemek zaman kaybı yapar ve gerçek sorunu gizleyebilir. Yalnızca kendiliğinden düzelebilecek hataları yeniden deneyin. mssql-python motor hata numarasını bir öznitelik olarak göstermediği için, istisna alt sınıfı ve metin driver_error ile sınıflandırın.

Bunları asla tekrar denemeyin (bunun yerine kodu veya yapılandırmayı düzeltin):

Koşul	Özel durum türü	driver_error metin	Düzelt
Geçersiz nesne adı (engine 208)	ProgrammingError	Base table or view not found	Tablo mevcut değil. Sorguyu düzeltin veya tabloyu oluşturun.
Geçersiz sütun adı (motor 207)	ProgrammingError	Column not found	Sütun mevcut değil. Şemayı kontrol et.
Yanlış sözdizimi (motor 102)	ProgrammingError	Syntax error or access violation	Soruyu düzeltin.
Giriş arızalandı (motor 18456)	OperationalError	Invalid authorization specification	Yanlış kimlik bilgileri. Bağlantı dizesini düzeltin.
Veritabanı açılamıyor (engine 4060)	OperationalError	Server rejected the connection	Veritabanı mevcut değil veya giriş için erişilebilir değil. Hedefi veya izinleri düzeltin.
Bağlantı havuzunun tükenmesi	OperationalError	(değişken)	Havuz kapasitesini artırın, bağlantıları hızlıca serbest bırakın veya eşzamanlılığı azaltın.
ConnectionStringParseError	Bağımsız	Yok	bağlantı dizesi anahtar kelimesinde yazım hatası. İpiyi düzelt.
Desteklenmeyen özellik	NotSupportedError	Optional feature not implemented	Alternatif bir yaklaşım kullanın.

Bunları her zaman tekrar deneyin (kendiliğinden çözülürler):

Koşul	Özel durum türü	driver_error metin
Deyim zaman aşımı	OperationalError	Timeout expired
Bağlantı zaman aşımı	OperationalError	Connection timeout expired
Bağlantı açılamadı	OperationalError	Client unable to establish connection
Ağ düşüşü	OperationalError	Communication link failure
İşlem sırasında bağlantı kesildi	OperationalError	Connection failure during transaction
Kilitlenme kurbanı (altyapı 1205)	OperationalError	Serialization failure
Belirsiz işlem durumu	OperationalError	Statement completion unknown
Azure SQL throttling (40197, 40501, 40613, 49918–49920)	ProgrammingError	Syntax error or access violation (motor numarası yalnızca ddbc_error içinde bulunur; is_azure_throttling kullanın)