mssql-python to sterownik języka Python firmy Microsoft dla programu SQL Server, usługi Azure SQL Database, usługi Azure SQL Managed Instance oraz bazy danych SQL w usłudze Microsoft Fabric. Korzysta z Direct Database Connectivity (DDBC), więc możesz się łączyć bez konieczności instalowania zewnętrznego menedżera sterowników. Sterownik obsługuje Python 3.10 lub nowszy i jest zgodny ze specyfikacją Python Database API 2.0, jednocześnie dodając usprawnienia przyjazne Python do codziennego rozwoju.
Wybieranie punktu początkowego
Plan bazowy produkcji dla Azure SQL
Użyj tego przykładu jako punktu wyjścia do produkcyjnego połączenia Azure SQL. Odczytuje konfigurację ze środowiska, uwierzytelnia się za pomocą tożsamości zarządzanej i umożliwia szyfrowanie Tabular Data Stream (TDS) 8.0. Ustawia również limity czasu logowania i wykonywania zapytań dla każdej instrukcji, ponawia próby po przejściowych błędach, stosując wykładniczo wydłużany czas między kolejnymi próbami (nowe połączenie w przypadku błędów połączenia, to samo połączenie w przypadku błędów zapytań, takich jak zakleszczenia), rejestruje wyniki i korzysta z menedżerów kontekstu do zwalniania zasobów.
Słowa kluczowe ConnectRetryCount i ConnectRetryInterval w parametrach połączenia włączają w SQL Server funkcję odporności nieaktywnego połączenia: sterownik automatycznie ponownie nawiązuje zerwane nieaktywne połączenie. To różni się od mechanizmu ponawiania na poziomie aplikacji w tym przykładzie, który ponawia zapytanie, gdy kończy się ono niepowodzeniem z powodu błędu przejściowego, takiego jak zakleszczenie lub przekroczenie limitu czasu zapytania. Te dwie rzeczy się uzupełniają, więc zachowaj oba.
import logging
import os
import time
import mssql_python
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s %(levelname)s %(name)s %(message)s",
)
logger = logging.getLogger("app")
# Transient errors that require a fresh connection to recover.
CONNECT_RETRY_ERRORS = frozenset({
"Timeout expired",
"Connection timeout expired",
"Client unable to establish connection",
"Communication link failure",
"Connection failure during transaction",
})
# Transient errors that leave the connection usable, such as a deadlock victim
# or a query timeout, so retry on the same connection.
QUERY_RETRY_ERRORS = frozenset({
"Serialization failure",
"Timeout expired",
})
def connect_with_retry(conn_str: str, max_attempts: int = 3, login_timeout_s: int = 5) -> mssql_python.Connection:
"""Open a connection, retrying transient failures with exponential backoff."""
for attempt in range(1, max_attempts + 1):
try:
conn = mssql_python.connect(
conn_str,
attrs_before={mssql_python.SQL_ATTR_LOGIN_TIMEOUT: login_timeout_s},
)
logger.info("connected on attempt %d/%d", attempt, max_attempts)
return conn
except mssql_python.OperationalError as exc:
if exc.driver_error not in CONNECT_RETRY_ERRORS or attempt == max_attempts:
logger.error("connect failed on attempt %d/%d: %s", attempt, max_attempts, exc.driver_error)
raise
delay = 2 ** (attempt - 1) # 1s, 2s, 4s
logger.warning(
"connect attempt %d/%d hit transient error %r; retrying in %ds",
attempt, max_attempts, exc.driver_error, delay,
)
time.sleep(delay)
def execute_with_retry(
conn: mssql_python.Connection,
sql: str,
*params,
max_attempts: int = 3,
query_timeout_s: int = 10,
) -> mssql_python.Cursor:
"""Run sql on an open connection and return the ready-to-fetch cursor.
Retries errors that leave the connection usable so callers don't wrap each
query in its own function. Pass query values as parameters. Retry only
idempotent statements; wrap writes in an explicit transaction.
"""
for attempt in range(1, max_attempts + 1):
cursor = mssql_python.Cursor(conn, timeout=query_timeout_s)
try:
cursor.execute(sql, *params)
if attempt > 1:
logger.info("query succeeded on attempt %d/%d", attempt, max_attempts)
return cursor
except mssql_python.OperationalError as exc:
cursor.close()
if exc.driver_error not in QUERY_RETRY_ERRORS or attempt == max_attempts:
logger.error("query failed on attempt %d/%d: %s", attempt, max_attempts, exc.driver_error)
raise
delay = 2 ** (attempt - 1) # 1s, 2s, 4s
logger.warning(
"query attempt %d/%d hit transient error %r; retrying in %ds",
attempt, max_attempts, exc.driver_error, delay,
)
time.sleep(delay)
raise RuntimeError("unreachable: the retry loop exits by return or raise")
def main() -> None:
# Read configuration from the environment; never hard-code secrets.
server = os.environ["SQL_SERVER"] # for example, myserver.database.windows.net
database = os.environ["SQL_DATABASE"] # for example, AdventureWorks
client_id = os.getenv("AZURE_CLIENT_ID") # set for a user-assigned managed identity
# Authenticate with the workload's managed identity over TDS 8.0 encryption.
# ConnectRetryCount/ConnectRetryInterval transparently reconnect a dropped
# idle connection; they don't replay a failed query.
conn_str = (
f"Server={server};"
f"Database={database};"
"Authentication=ActiveDirectoryMsi;"
"Encrypt=strict;"
"ConnectRetryCount=3;"
"ConnectRetryInterval=10;"
)
if client_id:
conn_str += f"UID={client_id};"
query = """
SELECT TOP 10
p.BusinessEntityID,
p.FirstName,
p.LastName
FROM Person.Person AS p
ORDER BY p.BusinessEntityID;
"""
try:
# Context managers close the cursor and connection automatically.
with connect_with_retry(conn_str) as conn:
with execute_with_retry(conn, query) as cursor:
for business_entity_id, first_name, last_name in cursor.fetchall():
print(f"{business_entity_id}\t{first_name}\t{last_name}")
except mssql_python.Error:
logger.exception("query failed")
raise
if __name__ == "__main__":
main()
Bardziej szczegółowe wskazówki dotyczące każdego z omówionych tu zagadnień można znaleźć w sekcjach uwierzytelnianie Microsoft Entra, buforowanie połączeń, szyfrowanie i certyfikaty, logika ponawiania oraz obsługa błędów.
Kluczowe funkcje
-
Zgodność z PEP 249: standardowe interfejsy
connect, cursor, execute i fetch*, oraz rozszerzenia zgodne z idiomami języka Python.
-
Bezpośrednia łączność z bazą danych (DDBC): Nie jest wymagany zewnętrzny menedżer sterowników. Zainstaluj
mssql-python i jesteś gotowy do połączenia.
-
Uwierzytelnianie za pomocą usługi Microsoft Entra ID: Wbudowana obsługa trybów uwierzytelniania, w tym tożsamości zarządzanych i nazw głównych usługi.
-
SQL Server i uwierzytelnianie systemu Windows: loginy SQL, Kerberos oraz logowanie jednokrotne systemu Windows (SSO) na obsługiwanych platformach.
-
Kopiowanie zbiorcze: Wysokowydajne wstawianie zbiorcze do ładowania dużych ilości danych z natywną obsługą protokołu TDS.
-
Obsługa natywnych typów danych: JSON, XML, typy przestrzenne, kolumny rzadkie, datetimeoffset oraz decimal/money z precyzyjną obsługą.
-
Integracja z Apache Arrow: Zbiory wyników bez kopiowania umożliwiające szybką wymianę danych z pandas, Polars i DuckDB.
-
Wzorce asynchroniczne: Używaj sterownika z aplikacjami opartymi na
asyncio oraz FastAPI, korzystając z obejść opartych na ThreadPoolExecutor. Zobacz wzorce asynchroniczne dotyczące integracji.
-
TLS domyślnie: szyfrowanie TLS i walidacja certyfikatów domyślnie włączone (za pomocą sterownika ODBC 18). Szyfrowanie TDS 8.0 jest dostępne po ustawieniu
Encrypt=strict.
Wprowadzenie
Praca z danymi
| Artykuł |
Opis |
|
Wykonywanie zapytań |
execute, executemany, partie wieloinstrukcyjne oraz zbiory wyników. |
|
Pobieranie danych |
fetchone, fetchmany, fetchall, oraz wzorce strumieniowe. |
|
Zapytania sparametryzowane |
Bezpiecznie wiąż parametry, aby zapobiec wstrzyknięciu SQL. |
|
Procedury składowane |
Wywołuj procedury, odczytuj parametry wyjściowe i zestawy wyników procesów. |
|
Zarządzanie kursorem |
Czas życia kursora, przewijanie i dostrajanie parametru arraysize. |
|
Obiekty wierszowe |
Uzyskuj dostęp do wierszy według indeksu, nazwy lub jako mapowań. |
|
Zarządzanie transakcjami |
Zatwierdzanie, wycofanie, punkty zapisu i poziomy izolacji. |
|
Dzielenie na strony |
Wzorce paginacji opartej na kluczach i paginacji z przesunięciem dla dużych zbiorów wyników. |
|
Obsługa błędów |
mssql_python.Error, DatabaseError, oraz strukturę błędów SQL Server. |
|
Logika ponowień |
Wykrywaj błędy przejściowe i ponawiaj próby z wykładniczo wydłużanymi odstępami. |
Typy danych i funkcje SQL Server
| Artykuł |
Opis |
|
Mapowania typów danych |
Tabela i reguły konwersji typu SQL Server-to-Python. |
|
Obsługa daty i godziny |
datetime, datetime2, datetimeoffset oraz kwestie związane ze strefami czasowymi. |
|
Typy danych dziesiętnych i walutowych |
Dokładne typy liczbowe i precyzja decimal.Decimal. |
|
Dane ciągowe i Unicode |
varchar, nvarchar, sortowania i strony kodowe. |
|
Obsługa NULL |
Logika trójwartościowa, strażnicy i pandas interoperacyjni. |
|
Dane binarne |
varbinary, image, oraz strumieniowanie dużych obiektów. |
|
Konwertery niestandardowe typu |
Rejestruj konwertery wejściowe i wyjściowe dla typów niestandardowych. |
|
Operacje kopiowania masowego |
Wysokowydajne wstawianie danych za pomocą interfejsu API kopiowania zbiorczego. |
|
Dane JSON |
Przechowuj, zapytuj i niszcz JSON z pomocą FOR JSON i OPENJSON. |
|
Dane XML |
Pracuj z typem xml danych, XPath i XQuery. |
|
Dane przestrzenne |
typy geometry oraz geography w Pythonie. |
|
Kolumny rozrzedłe |
Rzadkie kolumny i zestawy kolumn dla szerokich tabel. |
|
Odnajdywanie schematów |
Przeglądaj bazy danych, tabele, kolumny i indeksy. |
Wdrażanie i obsługa
Migracja do mssql-python
Odwołanie
| Artykuł |
Opis |
|
Cykl życia wsparcia |
Obsługiwane wersje Pythona i SQL Server oraz częstotliwość aktualizacji. |
|
Co nowego |
Historia wersji i najważniejsze informacje o wydaniu. |
Treści powiązane