Utilize o encaminhamento só de leitura e os grupos de disponibilidade

Os grupos de disponibilidade Always On oferecem alta disponibilidade para bases de dados SQL Server. Antes de ligar, o administrador da base de dados deve criar um ouvinte de grupo de disponibilidade e configurar o encaminhamento apenas de leitura no servidor. O driver mssql-python suporta ligações a grupos de disponibilidade através destas palavras-chave da cadeia de ligação:

• ApplicationIntent - Encaminhar ligações para réplicas secundárias apenas de leitura
• MultiSubnetFailover - Permitir tentativas de ligação paralela entre sub-redes para um failover mais rápido

Ligue-se ao listener do grupo de disponibilidade

Ligação básica do ouvinte

Ligue-se ao grupo de disponibilidade através do nome DNS do ouvinte em vez de um servidor específico:

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

Com especificação de porta

Especifique um número de porta quando o ouvinte usar uma porta não padrão:

conn = mssql_python.connect(
    "Server=ag-listener.contoso.com,1433;"  # Listener with port
    "Database=<database>;"
    "Authentication=ActiveDirectoryDefault;"
    "Encrypt=yes;"
)

Roteio de somente leitura

Ativar intenção de apenas leitura

Uso ApplicationIntent=ReadOnly para encaminhar para réplicas secundárias:

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

Verificar o encaminhamento

Confirme qual réplica processou a ligação e se é primária ou secundária:

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

Failover de várias sub-redes

Ativar a ativação pós-falha para múltiplas sub-redes

Para grupos de disponibilidade que abrangem múltiplas sub-redes:

conn = mssql_python.connect(
    "Server=ag-listener.contoso.com;"
    "Database=<database>;"
    "Authentication=ActiveDirectoryDefault;"
    "MultiSubnetFailover=yes;"
    "Encrypt=yes;"
)

Este cenário:

• Tenta estabelecer ligações a todos os endereços IP em paralelo.
• Reduz o tempo de failover em configurações de múltiplas sub-redes.
• Funciona melhor para todas as ligações a grupos de disponibilidade.

Padrões de arquitetura de ligação

Ligações de leitura e de escrita separadas

Mantenha conexões separadas para que o tráfego de leitura seja encaminhado para as secundárias e as operações de escrita para a primária:

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

Padrão de leitura após escrita

Escreva no primário e depois verifique os dados no secundário, tendo em conta o atraso de replicação:

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

Lidar com comutação automática

Resiliência da conexão

Repetir automaticamente as consultas quando um failover interrompe a ligação:

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

O ciclo de repetição reage apenas a códigos de erro transitórios. Falhas não transitórias, como erros de autenticação, erros de permissão ou erros de sintaxe de consulta, são apresentadas imediatamente porque a tentativa novamente não consegue resolvê-las.

Detetar mudança de função primária

Verifique a função atual da réplica e restabeleça a ligação se o primário tiver sido movido:

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

Cargas de trabalho de reporte

Transferir relatórios para o sistema secundário

Execute consultas de relatório contra uma réplica secundária para reduzir a carga sobre a primária:

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

Melhores práticas

Utilize sempre MultiSubnetFailover

Inclua MultiSubnetFailover=yes em todas as cadeias de ligação de grupos de disponibilidade para um failover mais rápido:

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

Associar a intenção à operação

Encaminhe cada operação para a réplica correta com base se esta lê ou escreve dados:

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

Manípulo secundário temporariamente indisponível

Tente aceder novamente ao secundário em caso de erros transitórios e, em seguida, recorra à réplica primária se o secundário continuar inacessível.

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

Conteúdo relacionado

• Cadeias de ligação
• Lógica de repetição
• Pool de conexões
• Grupos de disponibilidade Always On