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 友好的改进。

选择起点

Azure SQL的生产基线

本示例连接了带有托管标识和表数据流（TDS）8.0加密的Azure SQL。 它会读取环境变量中的连接设置，并设置独立的登录和查询超时。 每次重试都会重新调用 connect() ，而不是重用失败的连接对象。 上下文管理器在每次尝试后释放连接和光标。

仅将此辅助程序用于可安全重复执行的幂等读取查询。 不要用它来重放单个写入操作或部分完成的交易。

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

即使诊断中省略了错误编号，Azure 错误检查仍与标准的英文错误 40501 消息匹配。 它还会在存在时检查 ddbc_error 的已知数字。 匹配是尽力而为：本地化或格式不同的消息可能无法匹配。 更多信息请参见 瞬态连接错误。

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

关于本示例中每个问题的更深入指导，请参见 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：使用任一 本地 ODBC 提供商时，默认启用 TLS 加密和证书验证。 当你设置 Encrypt=strict时，可以使用 TDS 8.0 加密。

开始

文章 描述
安装 安装mssql-python并验证你的Python环境。
快速入门：连接mssql-python 连接到本地或测试的 SQL Server 实例，运行你的第一个查询。
快速入门：从Jupyter Notebook连接 在笔记本中使用 mssql-python 进行交互式数据探索。
快速入门：批量复制 将大型数据集通过批量复制API迁移到SQL Server。
快速入门：快速原型制作 快速构建小型脚本和概念验证。
快速启动：可重复部署 打包、配置并发布与SQL通信的Python应用程序。
Apache Arrow 快速入门 将查询结果导出为Apache Arrow表格，用于分析工作流。

配置和认证

文章 描述
mssql-odbc 提供程序（预览版） 选择预览原生提供商，查看支持的功能和已知问题。
连接字符串 连接字符串语法、常用关键词和示例。
程序化构建连接字符串 根据配置和机密安全地构建连接字符串。
连接管理 干净利落地打开、重复使用和关闭连接。
连接池 泳池调校、寿命和重复利用模式。
加密与证书 TLS加密模式、证书验证和TDS 8.0。
Microsoft Entra 身份验证 无密码认证适用于Azure SQL，包含管理身份、服务主体、交互式和设备代码流程。
安全最佳做法 参数化、机密管理、最低特权和加密。
可用性组 连接到 Always On 可用性组和只读副本。

处理数据

文章 描述
执行查询 execute、executemany、多语句批处理和结果集。
获取数据 fetchone、fetchmany、fetchall和流式模式。
参数化查询 安全绑定参数以防止SQL注入。
存储过程 调用过程，读取输出参数，处理结果集。
光标管理 光标寿命、滚动和数组大小调优。
行对象 通过索引、名称或映射形式访问行。
事务管理 提交、回滚、保存点和隔离级别。
分页 针对大型结果集的键集分页和偏移分页模式。
错误处理 mssql_python.Error、DatabaseError以及 SQL Server 错误结构。
重试逻辑 检测瞬态错误，并用指数回馈重试。

SQL Server 数据类型与特性

文章 描述
数据类型映射 SQL Server 到 Python 类型的表格和转换规则。
日期时间处理 datetime、datetime2、datetimeoffset以及时区注意事项。
十进制和货币类型 精确数值类型和 decimal.Decimal 精度。
字符串和Unicode数据 varchar、 nvarchar、排序和代码页。
NULL 处理 三值逻辑、哨兵和 pandas 互操作。
二进制数据 varbinary、image以及大型对象流式传输。
定制类型转换器 为自定义类型注册输入/输出转换器。
批量复制操作 使用批量复制 API 进行高吞吐量插入。
JSON 数据 使用 FOR JSON 和 OPENJSON 存储、查询和分解 JSON。
XML数据 使用 xml 数据类型、XPath 和 XQuery。
空间数据 geometry以及geography来自 Python 的类型。
稀疏列 宽表的稀疏列和列集。
架构发现 检查数据库、表格、列和索引。

与 Python 工具和框架集成

文章 描述
Apache Arrow 集成 将结果作为箭头表获取，用于零拷贝分析。
Pandas 集成 将查询结果加载到DataFrame中并写回。
Polars 集成 使用 Polars 配合 mssql-python 处理列式工作负载。
DuckDB 集成 同时查询 SQL Server 数据和本地 DuckDB 表。
FastAPI 集成 把 mssql-python 接入 FastAPI 服务。
测试和部署FastAPI应用程序 配置 FastAPI 服务的池化、错误、认证、测试和部署设置。
Flask 集成 在Flask应用中使用mssql-python。
异步模式 将mssql-python与 asyncio 线程池结合起来。
数据访问与分析模式 为光标访问、箭头提取、pandas、Polars 和对 SQL 数据的 DuckDB 分析选择正确的读取路径。
数据加载与移动模式 选择正确的写入路径，用于逐行插入、批量复制、MERGE upsert 操作、DataFrame 加载和导入 CSV 数据。

部署和操作

文章 描述
容器和本地开发 为连接SQL的Python应用设置Docker容器、开发容器和CI流水线。
性能优化 池优化、预处理语句、批大小和批量复制。
Troubleshooting 通过症状或错误信息查找指导。
安装与连接故障排除 解决安装、连接、容器和持续集成问题。
查询、数据与操作故障排除 解决查询、数据类型、性能、事务和批量复制问题。
模块配置 模块级设置、日志钩子和功能标志。

迁移到 mssql-python

文章 描述
从 pyodbc 迁移 将 pyodbc API 和连接字符串映射到 mssql-python。
从 pymssql 迁移 用mssql-python替换pymssql，同时保留行为。
从 SQLite 迁移 将本地SQLite工作负载迁移到SQL Server或Azure SQL。
从 PostgreSQL 迁移 Python 开发者使用 mssql-python 从 PostgreSQL 迁移到 SQL Server 的一站式指南

请求功能

要请求功能，请在 mssql-python GitHub 仓库中打开一个问题。

相关内容

  • 上次更新时间为