일시적 실패는 mssql-python 드라이버를 통해 SQL Server와 Azure SQL에 연결할 때 발생할 수 있는 일시적 오류입니다. 이러한 오류들은 종종 저절로 해결됩니다:
- 네트워크 연결 일시 중단.
- 서버 자원 제약.
- Azure SQL 제한
- 장애 전환 이벤트.
재시도 로직을 구현하면 특히 클라우드 호스팅 데이터베이스의 경우 애플리케이션 신뢰성이 향상됩니다.
설정이나 코딩 실수를 숨기기 위해 재시도를 사용하지 마세요. 데이터베이스가 누락되었거나, 잘못된 자격 증명이 발생하거나, 고갈된 연결 풀은 다시 시도할 필요가 없습니다.
일시적 오류 식별
mssql-python은 예외 시 SQL Server 엔진의 오류 번호를 속성으로 노출하지 않습니다. 대신 드라이버는 SQLSTATE 코드를 고정된 PEP 249 예외 하위 클래스 집합(OperationalError, ProgrammingError 등)과 driver_error 속성의 표준화된 영어 텍스트에 매핑합니다. 그 조합을 일시적 분류의 기초로 사용하세요.
신뢰할 수 있는 과도 신호
다음 SQLSTATE 값들은 다시 시도할 가치가 있는 조건으로 나타납니다 OperationalError . 오른쪽 열에는 운전자가 설정한 정확한 driver_error 문구가 나와 있습니다:
| SQLSTATE |
driver_error 본문 |
상태 |
|---|---|---|
HYT00 |
Timeout expired |
진술 수준 타임아웃. |
HYT01 |
Connection timeout expired |
연결 시간 초과. |
08001 |
Client unable to establish connection |
연결이 안 됐어. |
08S01 |
Communication link failure |
네트워크 끊김, 서버 초기화, TCP 실패. |
08007 |
Connection failure during transaction |
거래 도중 연결이 끊겼습니다. |
40001 |
Serialization failure |
교착 상태에 빠진 피해자. |
40003 |
Statement completion unknown |
거래 상태가 불확정입니다. |
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 제한(가능한 범위 내에서)
Azure SQL 스로틀링 오류(40197, 40501, 40613, 49918, 49919, 49920, 및 관련 코드)는 보통 SQLSTATE 42000로 전달되며, mssql-python에서는 이를 ProgrammingError에 매핑합니다. 엔진 오류 번호는 속성으로 표시되지 않기 때문에, 유일한 신호는 속성 내 ddbc_error 서버 메시지 텍스트뿐입니다.
워크로드가 Azure SQL을 대상으로 실행되고 스로틀링이 발생해 다시 시도해야 하는 경우, 알려진 번호를 확인하려면 ddbc_error을(를) 살펴보세요. 서버 측 텍스트의 형식이 안정적인 계약이 아니기 때문에 최선의 노력입니다:
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)
다시 시도하지 않을 내용
재시도 대신 빠르게 실패해야 하는 오류 예로는 무효 자격 증명(OperationalError 드라이버 텍스트 Invalid authorization specification포함), 누락 또는 접근 불가능한 데이터베이스, 문법 오류(),ProgrammingError 누락된 객체, 연결 풀 소진 등이 있습니다. 위의 함수는 is_transient_error 구성에 따라 이 모든 것을 배제한다.
기본 리트리 데코레이터
고정 지연으로 간단한 재시도
정해진 횟수만큼 일정한 간격을 두고 래핑된 함수를 재시도하는 데코레이터:
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()
지수 백오프
재시도 간 지연을 기하급수적으로 늘리고, 선택적 지터를 적용해 동시 재시도를 분산시키세요:
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()
연결 재시도 클래스
복원력 있는 연결 관리자
재시도와 자동 재연결을 모두 처리하는 연결 래퍼:
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 전용 처리
Azure 제한 처리
Azure SQL 제한 오류는 표준 일시적 오류보다 더 긴 지연과 더 많은 재시도가 필요합니다.
is_azure_throttling에서 재사용:
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
장애 조치(failover) 처리
Azure SQL 또는 가용성 그룹 장애 조치가 연결을 중단할 때 재연결하고 재시도하세요:
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"}
)
교착 상태 처리
교착 상태 시 재시도
교착 상태(오류 1205)는 일시적입니다. 교착 상태에서 벗어나기 위해 짧은 무작위 지연을 두고 다시 시도하세요. 재시도하면 즉각적인 실패를 해결하지만, 반복되는 교착 상태는 서버 측에서 조사해야 할 설계 문제를 나타냅니다. 근본 원인 분석과 해결에 대한 지침은 교착 상태 오류를 참조하세요.
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
구성이 있는 구조적 재시도
재시도 정책 클래스
재시도 구성을 데이터클래스에 캡슐화하여 다양한 연산에서 재사용하도록 합니다:
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)
회로 차단기 패턴
연속적인 오류를 추적하고 임계값에 도달하면 통화를 일시적으로 차단하여 연쇄 실패를 방지합니다:
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
관련 콘텐츠
설정 오류를 다시 시도하지 마세요
모든 오류가 일시적인 것은 아닙니다. 설정이나 코딩 오류를 다시 시도하면 시간 낭비가 되고 실제 문제를 가릴 수 있습니다. 저절로 해결될 수 있는 오류만 재시도하세요. mssql-python은 엔진 오류 번호를 속성으로 노출하지 않으니, 예외 서브클래스와 driver_error 텍스트별로 분류하세요.
이 모드들은 절대 다시 시도하지 마세요(대신 코드나 설정을 수정하세요):
| 상태 | 예외 유형 |
driver_error 본문 |
수정 |
|---|---|---|---|
| 유효하지 않은 객체 이름 (엔진 208) | ProgrammingError |
Base table or view not found |
테이블은 존재하지 않습니다. 쿼리를 수정하거나 테이블을 만드세요. |
| 유효하지 않은 열 이름 (엔진 207) | ProgrammingError |
Column not found |
해당 열이 존재하지 않습니다. 스키마를 확인해. |
| 잘못된 문법 (엔진 102) | ProgrammingError |
Syntax error or access violation |
쿼리를 고치세요. |
| 로그인 실패 (엔진 18456) | OperationalError |
Invalid authorization specification |
잘못된 자격증. 연결 문자열(연결 문자열)을 고치세요. |
| 데이터베이스를 열 수 없습니다 (엔진 4060) | OperationalError |
Server rejected the connection |
데이터베이스가 존재하지 않거나 로그인 시 접근할 수 없습니다. 목표나 권한 설정을 정하세요. |
| 연결 풀 고갈 | OperationalError |
(다양함) | 풀 용량을 늘리거나, 연결을 신속히 해제하거나, 동시성을 줄이세요. |
ConnectionStringParseError |
독립 실행형 | n/a | 연결 문자열 키워드에 오타가 있습니다. 줄을 고쳐. |
| 지원되지 않는 기능 | NotSupportedError |
Optional feature not implemented |
다른 방법을 사용하세요. |
항상 이 질문들을 다시 시도하세요 (스스로 해결됩니다):
| 상태 | 예외 유형 |
driver_error 본문 |
|---|---|---|
| 명령문 제한 시간 | OperationalError |
Timeout expired |
| 연결 시간 제한 | OperationalError |
Connection timeout expired |
| 연결이 열리지 않았습니다 | OperationalError |
Client unable to establish connection |
| 네트워크 연결 끊김 | OperationalError |
Communication link failure |
| 거래 도중 연결이 끊겼습니다 | OperationalError |
Connection failure during transaction |
| 교착 상태 피해자(엔진 1205) | OperationalError |
Serialization failure |
| 불확정 거래 상태 | OperationalError |
Statement completion unknown |
| Azure SQL throttling (40197, 40501, 40613, 49918–49920) | ProgrammingError |
Syntax error or access violation(엔진 번호는 ddbc_error에만 있습니다. is_azure_throttling를 사용하세요) |