Драйвер 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. Он также устанавливает тайм-ауты входа и выполнения отдельных запросов, повторяет попытки при временных сбоях с экспоненциально увеличивающейся задержкой (используя новое соединение при ошибках соединения и то же соединение при ошибках запроса, таких как взаимные блокировки), регистрирует результаты в журнале и полагается на менеджеры контекста для освобождения ресурсов.

Ключевые слова 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;"
        # 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:
        # 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: аутентификация, пул соединений, шифрование и сертификаты, логика повторного тестирования и обработка ошибок.

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

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

Статья Описание
Installation Установите mssql-python и проверьте вашу среду Python.
Быстрый старт: Подключитесь с помощью mssql-python Подключитесь к локальному или тестовому экземпляру SQL Server и выполните первый запрос.
Быстрый старт: подключитесь из Jupyter Notebook Используйте mssql-python внутри блокнота для интерактивного исследования данных.
Быстрый старт: массовое копиирование Переместите большие наборы данных в SQL Server с помощью API массового копирования.
Быстрый старт: Быстрое прототипирование Быстро создавайте небольшие скрипты и доказательства концепции.
Быстрый старт: повторяемые развертывания Упаковывайте, настраивайте и отправляйте Python-приложения, которые общаются с SQL.
Быстрый старт Apache Arrow Получите результаты запроса в виде таблиц Apache Arrow для аналитических рабочих процессов.

Настройка и аутентификация

Статья Описание
строки подключения Синтаксис строк соединения, распространённые ключевые слова и примеры.
Постройте строки соединения программным способом Безопасно формируйте строки подключения из конфигурации и секретов.
Управление подключениями Открывать, повторно использовать и корректно закрывать соединения.
Организация пулов соединений Настройка бассейна, сроки службы и паттерны повторного использования.
Шифрование и сертификаты режимы шифрования 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.
Интеграция с Flask Используйте mssql-python в приложениях Flask.
Асинхронные паттерны Сочетайте mssql-python с asyncio и пулами потоков.
Доступ к данным и аналитические паттерны Выберите правильный способ чтения для работы с курсором, извлечения в формате Arrow, а также для аналитики в pandas, Polars и DuckDB по данным SQL.
Загрузка данных и паттерны перемещения Выберите подходящий способ записи для вставки строк, массового копирования, MERGE операций upsert, загрузки DataFrame и импорта CSV.

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

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

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

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

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

  • Last updated on