Always Onの可用性グループはSQL Serverデータベースの高可用性を提供します。 接続前に、データベース管理者は 可用性グループリスナーを作成し 、サーバー上で 読み取り専用ルーティングを設定し なければなりません。 mssql-pythonドライバーは、以下の接続文字列キーワードを通じて可用性グループ接続をサポートしています:
-
ApplicationIntent- 読み取り専用の二次レプリカへの接続ルーティング -
MultiSubnetFailover- サブネット間での並列接続試行を有効化し、より高速なフェイルオーバーを実現する
可用性グループ リスナーに接続する
基本的なリスナー接続
特定のサーバーではなくリスナーDNS名を通じて可用性グループに接続します:
import mssql_python
# Connect via availability group listener
conn = mssql_python.connect(
"Server=ag-listener.contoso.com;"
"Database=<database>;"
"Authentication=ActiveDirectoryDefault;"
"Encrypt=yes;"
)
cursor = conn.cursor()
cursor.execute("SELECT @@SERVERNAME AS ServerName")
print(f"Connected to: {cursor.fetchval()}")
ポート仕様付き
リスナーがデフォルトでないポートを使用する場合、ポート番号を指定します:
conn = mssql_python.connect(
"Server=ag-listener.contoso.com,1433;" # Listener with port
"Database=<database>;"
"Authentication=ActiveDirectoryDefault;"
"Encrypt=yes;"
)
読み取り専用ルーティング
読み取り専用インテントを有効にする
ApplicationIntent=ReadOnlyを使ってセカンダリパリカへのルーティングを行います:
# Connect for read operations - routes to secondary
read_conn = mssql_python.connect(
"Server=ag-listener.contoso.com;"
"Database=<database>;"
"Authentication=ActiveDirectoryDefault;"
"ApplicationIntent=ReadOnly;"
"Encrypt=yes;"
)
# Connect for read-write operations - routes to primary
write_conn = mssql_python.connect(
"Server=ag-listener.contoso.com;"
"Database=<database>;"
"Authentication=ActiveDirectoryDefault;"
"ApplicationIntent=ReadWrite;" # Default
"Encrypt=yes;"
)
ルーティングの確認
どのレプリカが接続を担当したか、それがプライマリかセカンダリかを確認してください:
def check_replica_role(conn) -> str:
"""Check if connected to primary or secondary."""
cursor = conn.cursor()
cursor.execute("""
SELECT
@@SERVERNAME AS ServerName,
CASE
WHEN DATABASEPROPERTYEX(DB_NAME(), 'Updateability') = 'READ_WRITE'
THEN 'Primary'
ELSE 'Secondary'
END AS Role
""")
row = cursor.fetchone()
return f"{row.ServerName} ({row.Role})"
read_conn = mssql_python.connect(connection_string + "ApplicationIntent=ReadOnly;")
print(f"Read connection: {check_replica_role(read_conn)}")
write_conn = mssql_python.connect(connection_string + "ApplicationIntent=ReadWrite;")
print(f"Write connection: {check_replica_role(write_conn)}")
マルチサブネット フェールオーバー
マルチサブネットフェイルオーバーを有効にする
複数のサブネットにまたがる可用性グループの場合:
conn = mssql_python.connect(
"Server=ag-listener.contoso.com;"
"Database=<database>;"
"Authentication=ActiveDirectoryDefault;"
"MultiSubnetFailover=yes;"
"Encrypt=yes;"
)
この設定:
- すべてのIPアドレスへの接続を並行して試みます。
- マルチサブネット構成でのフェイルオーバー時間を短縮します。
- すべての可用性グループの接続で最も効果的です。
接続アーキテクチャのパターン
読み書き接続の分離
読み取りトラフィックはセカンダリに、書き込みはプライマリに送られるように別々の接続を維持しましょう:
class DatabaseConnections:
"""Manage separate connections for read and write operations."""
def __init__(self, listener: str, database: str):
self.base_conn_str = f"Server={listener};Database={database};Authentication=ActiveDirectoryDefault;Encrypt=yes;"
self._read_conn = None
self._write_conn = None
@property
def read_connection(self):
"""Get or create read-only connection (secondary replica)."""
if self._read_conn is None:
self._read_conn = mssql_python.connect(
self.base_conn_str + "ApplicationIntent=ReadOnly;MultiSubnetFailover=yes;"
)
return self._read_conn
@property
def write_connection(self):
"""Get or create read-write connection (primary replica)."""
if self._write_conn is None:
self._write_conn = mssql_python.connect(
self.base_conn_str + "ApplicationIntent=ReadWrite;MultiSubnetFailover=yes;"
)
return self._write_conn
def close(self):
if self._read_conn:
self._read_conn.close()
if self._write_conn:
self._write_conn.close()
# Usage
db = DatabaseConnections("ag-listener.contoso.com", "<database>")
# Queries go to secondary
cursor = db.read_connection.cursor()
cursor.execute("SELECT ProductID, Name, ListPrice FROM Production.Product")
products = cursor.fetchall()
# Writes go to primary
cursor = db.write_connection.cursor()
cursor.execute("INSERT INTO #Products (Name) VALUES (%(name)s)", {"name": "New Product"})
db.write_connection.commit()
db.close()
書き込み後読み取りパターン
プライマリに書き込み、その後セカンダリでデータを検証し、レプリケーション遅延を考慮します:
def create_order_and_verify(conn_manager, order_data: dict):
"""Create order on primary, verify on secondary with eventual consistency."""
# Write to primary
write_cursor = conn_manager.write_connection.cursor()
write_cursor.execute("""
INSERT INTO #Orders (CustomerID, Total) VALUES (%(cust)s, %(total)s);
SELECT SCOPE_IDENTITY();
""", order_data)
order_id = write_cursor.fetchval()
conn_manager.write_connection.commit()
# Wait for replication (in production, use more sophisticated approach)
import time
time.sleep(1)
# Verify on secondary
read_cursor = conn_manager.read_connection.cursor()
read_cursor.execute("SELECT * FROM #Orders WHERE OrderID = %(id)s", {"id": order_id})
if read_cursor.fetchone():
print(f"Order {order_id} replicated to secondary")
else:
print(f"Order {order_id} not yet replicated")
return order_id
フェールオーバーの処理
接続の弾力性
フェイルオーバーで接続が中断された場合、自動的に再試行クエリを行います:
import time
def execute_with_failover_retry(conn_str: str, query: str, params: dict,
max_retries: int = 3) -> list:
"""Execute query with automatic reconnection on failover."""
for attempt in range(max_retries + 1):
try:
conn = mssql_python.connect(conn_str)
cursor = conn.cursor()
cursor.execute(query, params)
results = cursor.fetchall()
conn.close()
return results
except mssql_python.OperationalError as e:
error_str = str(e)
# Check for failover-related errors
# Azure SQL transient error codes
# See: /azure/azure-sql/database/troubleshoot-common-errors-issues
if any(code in error_str for code in ["40613", "40197", "40501"]):
if attempt < max_retries:
print(f"Failover detected, retrying ({attempt + 1}/{max_retries})...")
time.sleep(5 * (attempt + 1)) # Exponential backoff
continue
raise
# Usage
results = execute_with_failover_retry(
"Server=<ag-listener>.contoso.com;Database=<database>;Authentication=ActiveDirectoryDefault;MultiSubnetFailover=yes;Encrypt=yes;",
"SELECT ProductID, Name, ListPrice FROM Production.Product WHERE ProductSubcategoryID = %(cat)s",
{"cat": 5}
)
リトライループは一時的なエラーコードにのみ反応します。 認証エラー、権限エラー、クエリ構文エラーなどの非一時的な失敗は、再試行では解決できないため即座に報告されます。
主要役割の変化を検出
現在のレプリカロールを確認し、プライマリが移動しているかどうかを再接続してください:
def is_primary(conn) -> bool:
"""Check if current connection is to primary replica."""
cursor = conn.cursor()
cursor.execute("""
SELECT DATABASEPROPERTYEX(DB_NAME(), 'Updateability') AS Updateability
""")
return cursor.fetchval() == 'READ_WRITE'
def ensure_primary(conn_str: str) -> mssql_python.Connection:
"""Ensure connection is to primary, reconnect if needed."""
conn = mssql_python.connect(conn_str + "ApplicationIntent=ReadWrite;")
if not is_primary(conn):
# Might happen during failover
conn.close()
time.sleep(2)
conn = mssql_python.connect(conn_str + "ApplicationIntent=ReadWrite;")
return conn
レポート作業量
レポートのセカンダリへのオフロード
プライマリの負荷を軽減するために、セカンダリプレプリカに対して報告クエリを実行します:
class ReportingService:
"""Service that runs reports on secondary replicas."""
def __init__(self, listener: str, database: str):
self.conn_str = (
f"Server={listener};"
f"Database={database};"
"Authentication=ActiveDirectoryDefault;"
"ApplicationIntent=ReadOnly;"
"MultiSubnetFailover=yes;"
"Encrypt=yes;"
)
def run_report(self, report_query: str, params: dict = None) -> list:
"""Execute report query on secondary replica."""
conn = mssql_python.connect(self.conn_str)
cursor = conn.cursor()
try:
cursor.execute(report_query, params or {})
return cursor.fetchall()
finally:
cursor.close()
conn.close()
def get_sales_summary(self, start_date, end_date) -> dict:
"""Run sales summary report."""
results = self.run_report("""
SELECT
YEAR(OrderDate) AS Year,
MONTH(OrderDate) AS Month,
COUNT(*) AS OrderCount,
SUM(TotalDue) AS TotalSales
FROM Sales.SalesOrderHeader
WHERE OrderDate BETWEEN %(start)s AND %(end)s
GROUP BY YEAR(OrderDate), MONTH(OrderDate)
ORDER BY Year, Month
""", {"start": start_date, "end": end_date})
return [
{"year": r.Year, "month": r.Month,
"orders": r.OrderCount, "sales": r.TotalSales}
for r in results
]
# Usage
reports = ReportingService("ag-listener.contoso.com", "sales_db")
summary = reports.get_sales_summary(date(2024, 1, 1), date(2024, 12, 31))
ベスト プラクティス
必ずMultiSubnetFailoverを使いましょう
フェールオーバーを高速化するために、すべての可用性グループ接続文字列に MultiSubnetFailover=yes を含める:
# Recommended for all AG connections
conn = mssql_python.connect(
"Server=ag-listener.contoso.com;"
"Database=<database>;"
"Authentication=ActiveDirectoryDefault;"
"MultiSubnetFailover=yes;" # Always include this
"Encrypt=yes;"
)
意図を操作に対応付ける
各操作をデータの読み書きかに基づいて正しいレプリカにルーティングします:
def get_appropriate_connection(operation: str, connections: DatabaseConnections):
"""Get connection appropriate for the operation type."""
read_operations = {"SELECT", "REPORT", "EXPORT", "ANALYTICS"}
if operation.upper() in read_operations:
return connections.read_connection
else:
return connections.write_connection
セカンダリが一時的に利用できない場合の処理
一時的なエラーでセカンダリを再試行し、セカンダリーが到達不能なままならプライマリレプリカに戻します。
import time
def query_with_fallback(conn_manager, query: str, params: dict, max_retries: int = 2):
"""Query the secondary, retrying transient errors before falling back to the primary."""
transient_codes = ["40613", "40197", "40501"]
for attempt in range(max_retries + 1):
try:
cursor = conn_manager.read_connection.cursor()
cursor.execute(query, params)
return cursor.fetchall()
except mssql_python.OperationalError as e:
# Re-raise failures that retrying can't fix, such as authentication,
# permission, or query syntax errors.
if not any(code in str(e) for code in transient_codes):
raise
# Retry the secondary for transient errors before giving up.
if attempt < max_retries:
time.sleep(2 * (attempt + 1))
continue
# Secondary still unavailable after retries: fall back to the primary.
print("Secondary unavailable, using primary")
cursor = conn_manager.write_connection.cursor()
cursor.execute(query, params)
return cursor.fetchall()