Microsoft Python SQL Server驱动 - mssql-python

mssql-python是Microsoft为SQL Server、Azure SQL 数据库、Azure SQL 托管实例和Microsoft Fabric中的SQL数据库开发的Python驱动程序。 它使用直接数据库连接（DDBC），所以你可以在不安装外部驱动管理器的情况下连接。 该驱动支持 Python 3.10 及更新版本，并符合 Python 数据库 API 规范 2.0，同时为日常开发增加了对 Python 友好的改进。

选择起点

• 要快速运行本地 SQL Server 示例，请从快速启动开始：连接 mssql-python 驱动。
• 要通过无密码认证连接 Azure SQL，首先使用 Microsoft Entra 认证和连接字符串。
• 要以交互方式探索数据，请从 通过 Jupyter Notebook 连接 或 快速原型设计 开始。
• 要高效移动大量数据，请参阅 批量复制操作 或 批量复制快速入门。
• 要从其他驱动迁移，可以选择 从 pyodbc 迁移、从 pymssql 迁移、从 SQLite 迁移，或从 PostgreSQL 迁移。

Azure SQL的生产基线

请以此示例作为面向生产环境的 Azure SQL 连接的起点。 它从环境中读取配置，通过管理身份进行身份验证，并启用表式数据流（TDS）8.0加密。 它还会设置登录超时和每条语句的查询超时，使用指数退避机制重试瞬时故障（连接错误时使用新连接，死锁等查询错误时使用同一连接），记录结果日志，并依赖上下文管理器来释放资源。

连接字符串中的 ConnectRetryCount 和 ConnectRetryInterval 关键字可启用 SQL Server 的空闲连接复原能力：驱动程序会透明地重新连接已断开的空闲连接。 这与本示例中的应用级重试不同，后者是重试因暂时错误（如死锁或查询超时）失败的 查询 。 两者互补，所以两者都保留。

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

关于本示例中每个问题的更深入指导，请参见 Microsoft Entra 认证、连接池、加密与证书、重试逻辑和错误处理。

主要功能

• PEP 249 合规性：标准 connect、 cursor、 execute和 fetch* 接口，以及 Python 扩展。
• 直接数据库连接（DDBC）：无需外部驱动程序管理器。 安装完 mssql-python 毕，你就可以连接了。
• Microsoft Entra ID 身份验证：内置对身份验证模式的支持，包括托管标识和服务主体。
• SQL Server 和 Windows 身份验证：支持平台上的 SQL 登录、Kerberos 和 Windows 单点登录（SSO）。
• 批量复制：支持原生 TDS 协议的高性能批量插入，适用于大规模数据加载。
• 原生数据类型支持： JSON、XML、空间、稀疏列、datetimeoffset、十进制/货币 ，并能精确处理。
• Apache Arrow 集成：零拷贝结果集，可与 pandas、Polars 和 DuckDB 进行快速数据交换。
• 异步模式：通过 ThreadPoolExecutor 变通方案，可将该驱动程序用于基于 asyncio 的应用程序和 FastAPI。 关于积分模式，请参见 异步模式 。
• 默认支持TLS：默认启用 TLS加密和证书验证 （通过ODBC驱动18）。 当你设置 Encrypt=strict时，可以使用 TDS 8.0 加密。

开始

配置和认证

处理数据

SQL Server 数据类型与特性

与 Python 工具和框架集成

部署和操作

迁移到 mssql-python

Reference

相关内容

• GitHub 上的 mssql-python
• PyPI 上的 mssql-python
• 发行说明
• 路线图
• Azure Python 开发人员中心
• Python 数据库 API 规范 2.0