一時的な障害は、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
フェールオーバーの処理
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)は一時的なものです。 デッドロックサイクルを破るために、短いランダム遅延で再挑戦してください。 再試行は即時の失敗を解決しますが、繰り返されるデッドロックは設計上の問題を示しており、サーバー側で調査すべきです。 根本原因の分析と解決に関する指針については、 Deadlock errorsを参照してください。
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 |
スタンドアロン | なし | 「接続文字列」キーワードにタイプミスがあります。 紐を直せ。 |
| サポートされていない機能 | 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を使います) |