Microsoft Python Driver for SQL Server - mssql-python

mssql-python is Microsoft's Python driver for SQL Server, Azure SQL Database, Azure SQL Managed Instance, and SQL database in Microsoft Fabric. It uses Direct Database Connectivity (DDBC), so you can connect without installing an external driver manager. The driver supports Python 3.10 or later and complies with the Python Database API Specification 2.0 while adding Python-friendly improvements for day-to-day development.

Choose your starting point

• To get a local SQL Server sample running quickly, start with Quickstart: Connect with the mssql-python driver.
• To connect to Azure SQL with passwordless authentication, start with Microsoft Entra authentication and Connection strings.
• To explore data interactively, start with Connect from a Jupyter Notebook or Rapid prototyping.
• To move large volumes of data efficiently, go to Bulk copy operations or the Bulk copy quickstart.
• To migrate from another driver, go to Migrate from pyodbc, Migrate from pymssql, Migrate from SQLite, or Migrate from PostgreSQL.

Production baseline for Azure SQL

Use this sample as a starting point for a production-oriented Azure SQL connection. It reads configuration from the environment, authenticates with managed identity, and enables Tabular Data Stream (TDS) 8.0 encryption. It also sets login and per-statement query timeouts, retries transient failures with exponential backoff (a fresh connection for connection errors, the same connection for query errors like deadlocks), logs outcomes, and relies on context managers to release resources.

The ConnectRetryCount and ConnectRetryInterval keywords in the connection string enable SQL Server idle connection resiliency: the driver transparently reconnects a dropped idle connection. That's distinct from the application-level retry in this sample, which retries a query that fails with a transient error such as a deadlock or query timeout. The two are complementary, so keep both.

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

For deeper guidance on each concern in this sample, see Microsoft Entra authentication, Connection pooling, Encryption and certificates, Retry logic, and Error handling.

Key features

• PEP 249 compliance: Standard connect, cursor, execute, and fetch* interfaces, plus Pythonic extensions.
• Direct Database Connectivity (DDBC): No external driver manager required. Install mssql-python and you're ready to connect.
• Microsoft Entra ID authentication: Built-in support for authentication modes, including managed identities and service principals.
• SQL Server and Windows authentication: SQL logins, Kerberos, and Windows single sign-on (SSO) on supported platforms.
• Bulk copy: High-performance bulk insert for large data loads with native TDS protocol support.
• Native data type support: JSON, XML, spatial, sparse columns, datetimeoffset, and decimal/money with precise handling.
• Apache Arrow integration: Zero-copy result sets for fast data interchange with pandas, Polars, and DuckDB.
• Async patterns: Use the driver with asyncio-based applications and FastAPI through ThreadPoolExecutor workarounds. See Async patterns for integration patterns.
• TLS by default: TLS encryption and certificate validation on by default (via ODBC Driver 18). TDS 8.0 encryption available when you set Encrypt=strict.

Get started

Configure and authenticate

Work with data

SQL Server data types and features

Integrate with Python tools and frameworks

Deploy and operate

Migrate to mssql-python

Reference

Related content

• mssql-python on GitHub
• mssql-python on PyPI
• Release notes
• Roadmap
• Azure Python Developer Center
• Python Database API Specification 2.0