Драйвер Microsoft Python для SQL Server — mssql-python

mssql-pythonявляется драйвером Python от Microsoft для SQL Server, База данных SQL Azure, Управляемый экземпляр SQL Azure и SQL Database в Microsoft Fabric. Он использует прямое подключение к базе данных (DDBC), так что можно подключаться без установки внешнего менеджера драйверов. Драйвер поддерживает Python 3.10 или более поздние версии и соответствует спецификации API Python Database Specification 2.0, добавляя улучшения, удобные для Python для повседневной разработки.

Выберите начальную точку

Производственные базовые показатели для Azure SQL

Этот пример подключается к Azure SQL с управляемой идентификацией и шифрованием Tabular Data Stream (TDS) 8.0. Он считывает настройки соединения из переменных среды и устанавливает отдельные тайм-ауты для входа и запросов. Каждая повторная попытка снова вызывает 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: аутентификация, пул соединений, шифрование и сертификаты, логика повторного тестирования и обработка ошибок.

Ключевые особенности

Начало работы

Статья Описание
Installation Установите mssql-python и проверьте вашу среду Python.
Быстрый старт: Подключитесь с помощью mssql-python Подключитесь к локальному или тестовому экземпляру SQL Server и выполните первый запрос.
Быстрый старт: подключитесь из Jupyter Notebook Используйте mssql-python внутри блокнота для интерактивного исследования данных.
Быстрый старт: массовое копиирование Переместите большие наборы данных в SQL Server с помощью API массового копирования.
Быстрый старт: Быстрое прототипирование Быстро создавайте небольшие скрипты и доказательства концепции.
Быстрый старт: повторяемые развертывания Упаковывайте, настраивайте и отправляйте Python-приложения, которые общаются с SQL.
Быстрый старт 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 to-Python и правила преобразования.
Обработка даты и времени datetime, datetime2, datetimeoffset и вопросы, связанные с часовыми поясами.
Десятичные и денежные типы Точные числовые типы и decimal.Decimal точность.
Строки и данные Unicode varchar, nvarchar, колляции и кодовые страницы.
Обработка с NULL Трёхзначная логика, стражи и панды взаимодействия.
Двоичные данные varbinary, image и потоковая передача больших объектов.
Пользовательские преобразователи типов Зарегистрируйте конвертеры ввода и вывода для пользовательских типов.
Операции массового копирования Высокопроизводительная вставка данных с помощью API массового копирования.
Данные JSON Храните, выполняйте запросы к JSON и разбирайте JSON с помощью FOR JSON и OPENJSON.
XML-данные Работайте с xml типом данных, XPath и XQuery.
Пространственные данные типы geometry и geography в Python.
Разреженные столбцы Редкие столбцы и наборы столбцов для широких таблиц.
Обнаружение схем Проверяйте базы данных, таблицы, столбцы и индексы.

Интегрируйте с инструментами и фреймворками Python

Статья Описание
Интеграция Apache Arrow Извлекайте результаты в виде таблиц Apache Arrow для аналитики без копирования данных.
Интеграция с pandas Загрузите результаты запросов в DataFrames и записывайте их обратно.
Интеграция поляров Используйте Polars с mssql-python для столбцевых нагрузок.
Интеграция с DuckDB Запрашивайте данные SQL Server вместе с локальными таблицами DuckDB.
Интеграция с FastAPI Подключите mssql-python к сервисам FastAPI.
Тестирование и развертывание приложений FastAPI Настройте пул, ошибки, аутентификацию, тесты и настройки развертывания для сервисов FastAPI.
Интеграция с Flask Используйте mssql-python в приложениях Flask.
Асинхронные паттерны Сочетайте mssql-python с asyncio и пулами потоков.
Доступ к данным и аналитические паттерны Выберите правильный способ чтения для работы с курсором, извлечения в формате Arrow, а также для аналитики в pandas, Polars и DuckDB по данным SQL.
Загрузка данных и паттерны перемещения Выберите подходящий способ записи для вставки строк, массового копирования, MERGE операций upsert, загрузки DataFrame и импорта CSV.

Развертывание и эксплуатация

Статья Описание
Контейнер и локальная разработка Настройте Docker-контейнеры, девконтейнеры и CI-конвейеры для Python-приложений, которые подключаются к SQL.
Настройка производительности Настройка пула, подготовленные выражения, размеры пакетов и массовое копирование.
Troubleshooting Найдите рекомендации по симптомам или сообщениям об ошибках.
Установка и устранение неполадок соединения Решите проблемы с установкой, подключением, контейнером и CI.
Запросы, данные и устранение неполадок операций Решайте вопросы запросов, типа данных, производительности, транзакций и массового копирования.
Конфигурация модуля Настройки на уровне модулей, логовые хуки и флаги функций.

Миграция на mssql-python

Статья Описание
Миграция с pyodbc Сопоставьте API pyodbc и строки подключения с mssql-python.
Переход с pymssql Замените pymssql на mssql-python, сохраняя при этом поведение.
Миграция с SQLite Переместите локальные рабочие нагрузки SQLite на SQL Server или Azure SQL.
Миграция из PostgreSQL Универсальное руководство для разработчиков Python, переходящих с PostgreSQL на SQL Server с помощью mssql-python.

Запрашивание функции

Чтобы запросить функцию, откройте проблему в репозитории GitHub на mssql-python.

Связанные материалы

  • Последнее обновление: