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

Tento vzor se propojuje s Azure SQL pomocí spravované identity a šifrování Tabular Data Stream (TDS) 8.0. Čte nastavení spojení z proměnných prostředí a nastavuje samostatné časové limity přihlášení a dotazů. Každý pokus volá connect() znovu místo opětovného použití objektu připojení, které selhalo. Správci kontextu po každém pokusu uvolní spojení a kurzor.

Tento pomocník používejte pouze pro idempotentní dotazy pro čtení, které lze bezpečně opakovat. Nepoužívejte ho k opakování jednotlivých zápisů nebo částečně dokončených transakcí.

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í.

Kontrola chyb Azure odpovídá standardní anglické zprávě error 40501, i když diagnostika vynechá číslo chyby. Také kontroluje ddbc_error na známá čísla, pokud jsou přítomna. Porovnání funguje jen orientačně: lokalizované nebo jinak formátované zprávy se nemusí shodovat. Pro více informací viz Přechodné chyby připojení.

import logging
import os
import re
import time

import mssql_python

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s %(levelname)s %(name)s %(message)s",
)
logger = logging.getLogger("app")

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

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")
_AZURE_40501_MESSAGE = "The service is currently busy. Retry the request after 10 seconds."


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 (
        _AZURE_40501_MESSAGE in ddbc_text
        or any(int(m) in AZURE_THROTTLING_ERRORS for m in _ERROR_NUMBER_RE.findall(ddbc_text))
    )


def query_with_retry(
    conn_str: str,
    sql: str,
    *params,
    max_attempts: int = 3,
    login_timeout_s: int = 5,
    query_timeout_s: int = 10,
) -> list:
    """Fetch an idempotent read query, reconnecting before each retry."""
    for attempt in range(1, max_attempts + 1):
        try:
            with mssql_python.connect(conn_str, timeout=login_timeout_s) as conn:
                conn.timeout = query_timeout_s
                with conn.cursor() as cursor:
                    cursor.execute(sql, *params)
                    rows = cursor.fetchall()
            logger.info("query succeeded on attempt %d/%d", attempt, max_attempts)
            return rows
        except mssql_python.Error as exc:
            azure_throttling = is_azure_throttling(exc)
            transient = (
                isinstance(exc, mssql_python.OperationalError)
                and exc.driver_error in TRANSIENT_DRIVER_ERRORS
            )
            if not (transient or azure_throttling) or attempt == max_attempts:
                logger.error("query failed on attempt %d/%d: %s", attempt, max_attempts, exc.driver_error)
                raise
            delay = (10 if azure_throttling else 1) * (2 ** (attempt - 1))
            logger.warning(
                "query attempt %d/%d hit transient error %r; retrying in %ds",
                attempt, max_attempts, exc.driver_error, delay,
            )
            time.sleep(delay)
    raise ValueError("max_attempts must be at least 1")


def main() -> None:
    # Read configuration from the environment; never hard-code secrets.
    server = os.environ["SQL_SERVER"]      # for example, <server>.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:
        rows = query_with_retry(conn_str, query)
        for business_entity_id, first_name, last_name in rows:
            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

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
mssql-odbc provider (Preview) Vyberte si nativního poskytovatele náhledu a zkontrolujte podporované funkce a známé problémy.
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.
Testujte a nasazujte aplikace FastAPI Nastavte pooling, chyby, autentizaci, testy a nastavení nasazení pro služby 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 Najděte pokyny podle symptomu nebo chybové zprávy.
Instalace a řešení problémů s připojením Vyřešte problémy s instalací, připojením, kontejnery a CI.
Dotazy, data a řešení problémů s provozem Vyřešit problémy s dotazy, datovým typem, výkonem, transakcemi a hromadným kopírováním.
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.

Žádost o funkci

Pro žádost o funkci otevřete problém v mssql-python GitHub repozitáři.

Související obsah

  • Poslední aktualizace v