Coba kembali logika dan ketahanan koneksi dengan mssql-python

Kegagalan sementara adalah kesalahan sementara yang dapat terjadi saat menyambungkan ke SQL Server dan Azure SQL melalui driver mssql-python. Kesalahan ini sering kali diselesaikan dengan sendirinya:

• Gangguan singkat pada konektivitas jaringan.
• Batasan sumber daya server.
• Pembatasan laju Azure SQL
• Kejadian failover

Menerapkan logika coba lagi meningkatkan keandalan aplikasi, terutama untuk database yang dihosting cloud.

Jangan gunakan percobaan ulang untuk menyembunyikan kesalahan konfigurasi atau pengkodean. Database yang hilang, kredensial buruk, atau kumpulan koneksi yang habis memerlukan perbaikan, bukan upaya lain.

Mengidentifikasi kesalahan sementara

mssql-python tidak mengekspos nomor kesalahan mesin SQL Server sebagai atribut pada pengecualian. Sebagai gantinya, driver memetakan kode SQLSTATE ke serangkaian subkelas pengecualian PEP 249 yang tetap (OperationalError, ProgrammingError, dan seterusnya) serta ke teks bahasa Inggris standar dalam atribut driver_error. Gunakan kombinasi itu sebagai dasar untuk klasifikasi sementara.

Sinyal transien yang andal

Nilai SQLSTATE berikut muncul sebagai OperationalError dan menunjukkan kondisi yang layak untuk dicoba kembali. Kolom kanan menunjukkan teks yang tepat driver_error yang ditetapkan oleh pengemudi:

SQLSTATE	driver_error teks	Condition
HYT00	Timeout expired	Batas waktu pada tingkat pernyataan.
HYT01	Connection timeout expired	Batas waktu koneksi.
08001	Client unable to establish connection	Tidak dapat membuka koneksi.
08S01	Communication link failure	Penurunan jaringan, reset server, kegagalan TCP.
08007	Connection failure during transaction	Koneksi terputus di tengah transaksi.
40001	Serialization failure	Korban kebuntuan.
40003	Statement completion unknown	Status transaksi yang tidak ditentukan.

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

Pembatasan laju Azure SQL (sebisa mungkin)

Kesalahan throttling Azure SQL (40197, 40501, 40613, 49918, 49919, 49920, dan kode terkait) biasanya muncul dengan SQLSTATE 42000, yang dipetakan oleh mssql-python ke ProgrammingError. Nomor kesalahan mesin tidak muncul sebagai atribut, jadi satu-satunya sinyal adalah teks pesan server dalam ddbc_error atribut.

Jika beban kerja Anda berjalan di Azure SQL dan Anda perlu mencoba ulang saat terjadi throttling, periksa ddbc_error untuk nomor yang dimaksud. Ini adalah upaya terbaik karena format teks sisi server bukanlah kontrak yang stabil:

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)

Apa yang tidak perlu diulang

Contoh kesalahan yang seharusnya gagal dengan cepat alih-alih mencoba lagi termasuk kredensial yang tidak valid (OperationalError dengan teks Invalid authorization specificationdriver), database yang hilang atau tidak dapat diakses, kesalahan sintaks (ProgrammingError), objek yang hilang, dan kehabisan kumpulan koneksi. Fungsi is_transient_error di atas mengecualikan semuanya secara bawaan.

Dekorator coba ulang dasar

Coba ulang sederhana dengan penundaan tetap

Dekorator yang mencoba ulang fungsi yang dibungkus beberapa kali dengan penundaan konstan:

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

Backoff eksponensial

Tingkatkan penundaan antara percobaan ulang secara eksponensial dengan jitter opsional untuk menyebarkan percobaan ulang bersamaan:

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

Kelas percobaan ulang koneksi

Manajer koneksi yang tangguh

Pembungkus koneksi yang menangani coba ulang dan penyambungan ulang otomatis:

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

Penanganan khusus Azure SQL

Menangani pembatasan laju Azure

Kesalahan pelambatan Azure SQL memerlukan penundaan yang lebih lama dan lebih banyak percobaan ulang daripada kesalahan sementara biasa. Gunakan kembali is_azure_throttling dari bagian Identifikasi kesalahan transien:

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

Menangani proses failover

Sambungkan kembali dan coba lagi saat failover Azure SQL atau grup ketersediaan mengganggu koneksi:

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

Penanganan kebuntuan

Coba lagi saat kebuntuan

Deadlock (error 1205) bersifat sementara. Coba lagi dengan penundaan acak singkat untuk memutus siklus kebuntuan. Mencoba ulang mengatasi kegagalan saat itu, tetapi deadlock yang berulang menunjukkan adanya masalah desain yang harus Anda selidiki pada sisi server. Untuk panduan tentang menganalisis dan menyelesaikan akar penyebabnya, lihat Kesalahan kebuntuan.

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

Percobaan ulang terstruktur dengan konfigurasi

Coba lagi kelas kebijakan

Merangkum konfigurasi coba lagi dalam kelas data untuk digunakan kembali di berbagai operasi:

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)

Pola pemutus sirkuit

Cegah kegagalan bertingkat dengan melacak kesalahan berturut-turut dan memblokir panggilan sementara saat ambang batas tercapai:

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

Konten terkait

• Penanganan kesalahan
• Pengumpulan koneksi
• Troubleshooting
• Ketahanan koneksi Azure SQL

Jangan mencoba lagi kesalahan konfigurasi

Tidak semua kesalahan bersifat sementara. Mencoba kembali kesalahan konfigurasi atau pengkodean membuang waktu dan dapat menutupi masalah sebenarnya. Hanya coba lagi kesalahan yang mungkin diselesaikan dengan sendirinya. Karena mssql-python tidak mengekspos nomor kesalahan mesin sebagai atribut, klasifikasikan berdasarkan subkelas pengecualian ditambah driver_error teks.

Jangan pernah mencoba lagi ini (perbaiki kode atau konfigurasi sebagai gantinya):

Condition	Jenis pengecualian	driver_error teks	Perbaiki
Nama objek tidak valid (engine 208)	ProgrammingError	Base table or view not found	Tabel tidak ada. Perbaiki kueri atau buat tabel.
Nama kolom tidak valid (mesin 207)	ProgrammingError	Column not found	Kolom tidak ada. Periksa skemanya.
Sintaks salah (mesin 102)	ProgrammingError	Syntax error or access violation	Perbaiki kueri.
Login gagal (mesin 18456)	OperationalError	Invalid authorization specification	Kredensial salah. Perbaiki string koneksi.
Tidak dapat membuka database (mesin 4060)	OperationalError	Server rejected the connection	Database tidak ada atau tidak dapat diakses untuk login ini. Perbaiki target atau izin.
Kelelahan kumpulan koneksi	OperationalError	(bervariasi)	Tingkatkan kapasitas kumpulan, lepaskan koneksi segera, atau kurangi konkurensi.
ConnectionStringParseError	Mandiri	n/a	Kesalahan pengetikan pada kata kunci string koneksi. Perbaiki string.
Fitur yang tidak didukung	NotSupportedError	Optional feature not implemented	Gunakan pendekatan alternatif.

Selalu coba lagi ini (mereka menyelesaikannya sendiri):

Condition	Jenis pengecualian	driver_error teks
Batas waktu pernyataan	OperationalError	Timeout expired
Batas waktu koneksi	OperationalError	Connection timeout expired
Tidak dapat membuka koneksi	OperationalError	Client unable to establish connection
Penurunan jaringan	OperationalError	Communication link failure
Koneksi terputus di tengah transaksi	OperationalError	Connection failure during transaction
Korban kebuntuan (mesin 1205)	OperationalError	Serialization failure
Status transaksi yang tidak ditentukan	OperationalError	Statement completion unknown
Pembatasan Azure SQL (40197, 40501, 40613, 49918–49920)	ProgrammingError	Syntax error or access violation (nomor mesin hanya ada di ddbc_error; gunakan is_azure_throttling)