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;"
)
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
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. |
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. |
Odkaz
Související obsah