Ovladač Microsoft Pythonu pro SQL Server – mssql-python

mssql-python je pythonový ovladač společnosti Microsoft pro SQL Server, Azure SQL Database, Azure SQL Managed Instance a databázi SQL v prostředí Microsoft Fabric. Používá Direct Database Connectivity (DDBC), takže se můžete připojit bez instalace externího správce ovladačů. Ovladač podporuje Python 3.10 nebo novější a splňuje specifikaci Python Database API 2.0, přičemž přidává vylepšení přátelská k Python pro každodenní vývoj.

Výběr výchozího bodu

Směrný plán výroby pro Azure SQL

Použijte tento vzor jako výchozí bod pro produkčně orientované Azure SQL připojení. Čte konfiguraci z prostředí, autentizuje se spravovanou identitou a umožňuje šifrování Tabular Data Stream (TDS) 8.0. Také nastavuje časové limity pro přihlášení i pro jednotlivé dotazy, při přechodných selháních opakuje pokusy s exponenciálně rostoucí prodlevou (nové připojení při chybách připojení, stejné připojení při chybách dotazu, například při deadlocku), zaznamenává výsledky a k uvolňování prostředků využívá správce kontextu.

Klíčová slova ConnectRetryCount a ConnectRetryInterval v připojovacím řetězci umožňují v SQL Serveru odolnost nečinných připojení: ovladač transparentně znovu připojí odpojené nečinné připojení. To se liší od opakování na úrovni aplikace v této ukázce, které zopakuje dotaz, pokud selže kvůli přechodné chybě, například zablokování nebo vypršení časového limitu dotazu. Oba jsou vzájemně doplňující, takže si nechte obojí.

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;"
        # Parallel dials to all resolved IPs; safe on single-IP targets.
        "MultiSubnetFailover=Yes;"
    )
    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()

Pro podrobnější informace o každém problému v tomto vzoru viz Microsoft Entra autentizace, poolování spojení, šifrování a certifikáty, logika opakování a zpracování chyb.

Klíčové funkce

  • Soulad s PEP 249: Standardní rozhraní connect, cursor, execute a fetch* a navíc pythonovská rozšíření.
  • Přímé připojení k databázi (DDBC): Externí správce ovladačů není potřeba. Nainstalujte mssql-python a můžete se připojit.
  • Autentizace Microsoft Entra ID: Vestavěná podpora autentizačních režimů, včetně spravovaných identit a principů služeb.
  • SQL Server a Windows authentication: SQL přihlašování, Kerberos a Windows single sign-on (SSO) na podporovaných platformách.
  • Hromadné kopírování: Vysoce výkonné hromadné vkládání pro načítání velkých objemů dat s nativní podporou protokolu TDS.
  • Podpora nativních datových typů: JSON, XML, prostorová data, řídké sloupce, datetimeoffset a datové typy decimal/money s přesným zpracováním.
  • Integrace Apache Arrow: Sady výsledků bez kopírování dat pro rychlou výměnu dat s pandas, Polars a DuckDB.
  • Asynchronní vzorce: Používejte ovladač s aplikacemi založenými na asyncio a s FastAPI prostřednictvím obcházek s využitím ThreadPoolExecutoru. Viz Asynchronní vzory pro integrační vzory.
  • TLS ve výchozím nastavení: Šifrování TLS a ověřování certifikátů je zapnuto ve výchozím nastavení (přes ODBC ovladač 18). Šifrování TDS 8.0 je dostupné, když nastavíte Encrypt=strict.

Začínáme

Článek Popis
Installation Nainstalujte mssql-python a ověřte své prostředí Python.
Rychlý start: Připojte se k mssql-python Připojte se k lokální nebo testovací instanci SQL Server a spusťte první dotaz.
Rychlý začátek: Připojte se z Jupyter Notebook Používejte mssql-python v notebooku pro interaktivní průzkum dat.
Rychlý začátek: Hromadná kopie Přesouvej velké datové sady do SQL Server pomocí hromadného kopírovacího API.
Rychlý začátek: Rychlé prototypování Rychle vytvářejte malé skripty a důkazy konceptu.
Rychlý start: Opakovatelná nasazení Balíčky, konfigurace a distribuce Python aplikací, které komunikují s SQL.
Rychlý start Apache Arrow Načítejte výsledky dotazů ve formátu tabulek Apache Arrow pro analytické pracovní postupy.

Konfigurujte a autentizujte

Článek Popis
Připojovací řetězce Syntaxe spojovacích řetězců, běžná klíčová slova a příklady.
Programově budujte spojovací řetězce Bezpečně sestavujte řetězce připojení z konfigurace a tajných údajů.
Správa připojení Otevři, znovu použij a uzavírej spoje čistě.
Sdružování připojení Ladění bazénu, životnost a vzory opětovného použití.
Šifrování a certifikáty TLS šifrovací režimy, ověřování certifikátů a TDS 8.0.
Ověřování Microsoft Entra Ověřování bez hesla pro Azure SQL s využitím toků spravované identity, instančního objektu služby, interaktivního toku a toku kódu zařízení.
Osvědčené postupy zabezpečení Parametrizace, správa tajných kódů, nejnižší oprávnění a šifrování.
Skupiny dostupnosti Připojte se k dostupnostním skupinám Always On a replikám pouze pro čtení.

Práce s daty

Článek Popis
Provádění dotazů execute, executemany, dávky s více příkazy a sady výsledků.
Získávání dat fetchone, fetchmany, fetchall, a proudové vzory.
Parametrizované dotazy Bezpečně navázejte parametry, abyste zabránili SQL injekci.
Uložené procedury Volejte procedury, čtěte výstupní parametry a zpracovávejte sady výsledků.
Správa kurzorů Životnost kurzoru, posouvání a ladění velikosti pole.
Řádkové objekty Přistupujte k řádkům podle indexu, názvu nebo jako mapování.
Správa transakcí Potvrzení, vrácení změn, body uložení a úrovně izolace.
Stránkování Vzory stránkování pomocí klíče a offsetu pro velké sady výsledků.
Zpracování chyb mssql_python.Error, DatabaseError, a chybová struktura SQL Server.
Logika opakování Detekujte přechodné chyby a zkuste to znovu s exponenciálním ústupem.

Typy dat a funkce SQL Server

Článek Popis
Mapování datových typů Tabulky a konverzní pravidla typu SQL Server na Python.
Zpracování data a času datetime, datetime2, datetimeoffset a aspekty časových pásem.
Desetinné a peněžní typy Přesné číselné typy a decimal.Decimal přesnost.
Řetězcová a Unicode data varchar, nvarchar, řazení a kódové stránky.
Zpracování hodnoty NULL Trojhodnotová logika, sentinely a interoperabilita s pandas.
Binární data varbinary, image a streamování velkých objektů.
Vlastní převodníky typů Registrujte vstupní a výstupní převodníky pro vlastní typy.
Operace hromadného kopírování Vysoce propustné vkládání pomocí rozhraní API pro hromadné kopírování.
JSON data Uložit, dotazovat a skartovat JSON pomocí FOR JSON a OPENJSON.
XML data Pracuj s xml datovým typem, XPath a XQuery.
Prostorová data geometry a geography typy z Pythonu.
Řídké sloupce Řídké sloupce a sady sloupců pro široké tabulky.
Zjišťování schématu Prohlížejte databáze, tabulky, sloupce a indexy.

Integrace s nástroji a frameworky v Python

Článek Popis
Integrace Apache Arrow Načítání výsledků jako šipkové tabulky pro analytiku bez kopírovaní.
integrace s pandas Načte výsledky dotazů do DataFrames a zapisujte je zpět.
Integrace s Polars Používejte Polars s mssql-python pro sloupcově orientované úlohy.
Integrace DuckDB Dotazujte se na data SQL Serveru společně s místními tabulkami DuckDB.
Integrace s FastAPI Připojte mssql-python do služeb FastAPI.
Integrace s baňkou Používejte mssql-python v aplikacích ve Flasku.
Asynchronní vzory Kombinujte mssql-python s asyncio a fondy vláken.
Přístupové a analytické vzorce přístupu k datům Vyberte správný způsob čtení dat pro přístup přes kurzor, extrakci do Apache Arrow a analýzu v knihovnách pandas, Polars a DuckDB nad daty SQL.
Načítání dat a pohybové vzory Vyberte správnou cestu zápisu pro vkládání řádků, hromadné kopírování, MERGE operace typu upsert, načítání DataFrame a import CSV.

Nasazení a provoz

Článek Popis
Kontejner a místní vývoj Nastavte Docker kontejnery, devkontejnery a CI pipeline pro Python aplikace, které se připojují ke SQL.
Ladění výkonu Ladění v poolu, připravené výpisy, velikosti šarží a hromadné kopie.
Troubleshooting Běžné chyby, logování a diagnostika certifikátů.
Konfigurace modulu Nastavení na úrovni modulů, logovací háčky a feature flagy.

Migrujte na mssql-python

Článek Popis
Migrujte z pyodbc Namapujte rozhraní API pyodbc a připojovací řetězce na mssql-python.
Migrace z pymssql Nahraďte pymssql nástrojem mssql-python a zachovejte chování.
Migrujte ze SQLite Přesuňte lokální SQLite workloads na SQL Server nebo Azure SQL.
Migrace z PostgreSQL Komplexní průvodce pro Python vývojáře, kteří přecházejí z PostgreSQL na SQL Server s mssql-python.

Související obsah

  • Last updated on