Используйте mssql-python с SQLAlchemy

SQLAlchemy — самый широко используемый набор инструментов для Python ORM и баз данных. Начиная со SQLAlchemy 2.1.0b2, встроенный диалект драйвера mssql-python позволяет использовать SQLAlchemy ORM и Core с Microsoft SQL и База данных SQL Azure.

Important

Диалект mssql-python был добавлен в SQLAlchemy 2.1.0b2 (выпущен 16 апреля 2026 года). SQLAlchemy 2.1 в настоящее время является предрелизной серией и не рекомендуется для использования в производстве. Перед обновлением с SQLAlchemy 2.0 поймите:

  • API могут измениться до финального стабильного релиза (2.1 GA)
  • Тщательно протестируйте свою нагрузку перед развертыванием
  • Используйте стабильную версию SQLAlchemy 2.0.x для производственных систем до выхода общедоступной версии 2.1
  • Прикрепите зависимость к конкретной версии (например, sqlalchemy==2.1.0b2) вместо использования диапазонов версий

См. раздел «Известные ограничения » для подробностей о том, когда использовать версии до релиза.

Необходимые условия

  • Python 3.10 или более поздней версии. SQLAlchemy 2.1 прекратила поддержку Python 3.9 и более ранних версий.
  • Пакеты mssql-python и sqlalchemy (2.1.0b2 или новее).

Примеры в этой статье используют базу данных AdventureWorksLT . Если у вас не установлен AdventureWorksLT, посмотрите примеры баз данных AdventureWorks.

Установите пре-релиз

Поскольку SQLAlchemy 2.1 находится в бета-версии, pip install sqlalchemy по умолчанию устанавливается последний стабильный релиз 2.0.x. Установите пререлиз явно:

pip install mssql-python "sqlalchemy>=2.1.0b2"

Проверьте установленную версию:

import sqlalchemy
print(sqlalchemy.__version__)  # Should show 2.1.0b2 or later

URL-адреса соединений

Диалект mssql-python mssql+mssqlpython используется в качестве схемы URL. Общий формат:

mssql+mssqlpython://<username>:<password>@<host>:<port>/<database>

Проверка подлинности SQL

Для SQL-аутентификации указывайте имя пользователя и пароль в URL соединения:

from sqlalchemy import create_engine

# Replace <password> with your actual password. Avoid using the sa account in production.
engine = create_engine(
    "mssql+mssqlpython://dbuser:<password>@localhost:1433/<database>"
)

аутентификация Microsoft Entra

Для аутентификации Microsoft Entra используйте пустое имя пользователя и authentication параметр запроса:

from sqlalchemy import create_engine

engine = create_engine(
    "mssql+mssqlpython://@<server>.database.windows.net/<database>"
    "?authentication=ActiveDirectoryDefault&encrypt=yes"
)

Замечание

ActiveDirectoryDefault использует DefaultAzureCredential, который последовательно пробует несколько поставщиков учетных данных. Первое соединение может быть медленным, потому что SDK идёт по цепочке, пока не найдёт работающего провайдера. В продакшене, если вы знаете, какой тип учетных данных использует ваша среда, укажите его напрямую (например, ActiveDirectoryMSI для управляемой идентичности), чтобы избежать цепной ходьбы. Дополнительные сведения см. в разделе проверки подлинности Microsoft Entra.

Программное построение URL

Используйте sqlalchemy.engine.URL.create для предотвращения ручного кодирования URL:

from sqlalchemy.engine import URL

url = URL.create(
    "mssql+mssqlpython",
    username="dbuser",
    password="<password>",
    host="localhost",
    port=1433,
    database="<database>",
)
engine = create_engine(url)

Определить модели ORM

Используйте декларативное отображение SQLAlchemy, чтобы определить модели, соответствующие таблицам Microsoft SQL.

from datetime import datetime
from decimal import Decimal

from sqlalchemy import Identity, String, Numeric, Integer, DateTime, func
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column


class Base(DeclarativeBase):
    pass


class Product(Base):
    __tablename__ = "Product"
    __table_args__ = {"schema": "SalesLT"}

    product_id: Mapped[int] = mapped_column(
        "ProductID", Integer, Identity(), primary_key=True
    )
    name: Mapped[str] = mapped_column("Name", String(50))
    product_number: Mapped[str] = mapped_column("ProductNumber", String(25))
    color: Mapped[str | None] = mapped_column("Color", String(15))
    list_price: Mapped[Decimal] = mapped_column("ListPrice", Numeric(19, 4))
    standard_cost: Mapped[Decimal] = mapped_column("StandardCost", Numeric(19, 4))
    size: Mapped[str | None] = mapped_column("Size", String(5))
    product_category_id: Mapped[int | None] = mapped_column("ProductCategoryID", Integer)
    sell_start_date: Mapped[datetime] = mapped_column("SellStartDate", DateTime)
    modified_date: Mapped[datetime] = mapped_column(
        "ModifiedDate", DateTime, server_default=func.getdate()
    )

Tip

Microsoft SQL использует IDENTITY для автоинкрементируемых столбцов. SQLAlchemy автоматически отображает это для целочисленных столбцов первичных ключей. Приведённое выше явное Identity() указание необязательно, если только вам не нужно управлять начальными и инкрементными значениями.

CRUD-операции

Следующие примеры показывают, как вставлять, задавать запросы, обновлять и удалять строки с помощью сессии ORM. В каждом примере повторно используется new_id — ProductID, возвращаемое при вставке строки. Чтобы выполнить все четыре операции вместе, см. полный пример.

Создание сеанса

Создайте сессию для выполнения операций внутри транзакции:

from sqlalchemy.orm import Session

with Session(engine) as session:
    # Use session for queries and modifications
    pass

Для приложений, создающих множество сессий, используйте sessionmaker:

from sqlalchemy.orm import sessionmaker

SessionLocal = sessionmaker(bind=engine)

Вставка строк

Добавьте новый продукт, зафиксируйте сессию и зафиксируйте сгенерированный ProductID материал для следующих примеров:

from datetime import datetime

with Session(engine) as session:
    product = Product(
        name="Classic Road Bike",
        product_number="BK-C001",
        color="Red",
        list_price=Decimal("1299.99"),
        standard_cost=Decimal("749.99"),
        sell_start_date=datetime(2026, 1, 1),
        product_category_id=6,
    )
    session.add(product)
    session.commit()

    new_id = product.product_id
    print(f"Inserted ProductID: {new_id}")

Замечание

В SalesLT.Product и Name, и ProductNumber имеют уникальные ограничения. Если вы запускаете эту вставку несколько раз, сначала измените эти значения или удалите предыдущую строку. Полный пример удаляет созданную строку, чтобы она могла выполняться многократно.

Строки запросов

Получите одну строку по первичному ключу или используйте select() для отфильтрованных запросов:

from sqlalchemy import select

with Session(engine) as session:
    # Single row by primary key (new_id is from the insert example)
    product = session.get(Product, new_id)
    if product:
        print(f"{product.name}: ${product.list_price}")

    # Filtered query
    stmt = select(Product).where(Product.list_price < 500).order_by(Product.name)
    products = session.scalars(stmt).all()
    for p in products:
        print(f"{p.name}: ${p.list_price}")

Обновление строк

Измените поле на существующей строке и зафиксируйте:

with Session(engine) as session:
    product = session.get(Product, new_id)
    if product:
        product.list_price = Decimal("1349.99")
        session.commit()

Удаление строк

Удалите строку и сделайте коммит:

with Session(engine) as session:
    product = session.get(Product, new_id)
    if product:
        session.delete(product)
        session.commit()

Полный пример

В предыдущих разделах каждая работа была показана отдельно. Этот раздел объединяет их в один самостоятельный скрипт, который можно копировать, запускать и запускать заново.

Создайте файл с именем crud.py и добавьте следующий код. Замените данные create_engine соединения на свои (см. URL соединений):

from datetime import datetime
from decimal import Decimal

from sqlalchemy import create_engine, Identity, String, Numeric, Integer, DateTime, func
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, Session

# Replace <password> and <database> with your connection details.
engine = create_engine(
    "mssql+mssqlpython://dbuser:<password>@localhost:1433/<database>"
)


class Base(DeclarativeBase):
    pass


class Product(Base):
    __tablename__ = "Product"
    __table_args__ = {"schema": "SalesLT"}

    product_id: Mapped[int] = mapped_column(
        "ProductID", Integer, Identity(), primary_key=True
    )
    name: Mapped[str] = mapped_column("Name", String(50))
    product_number: Mapped[str] = mapped_column("ProductNumber", String(25))
    color: Mapped[str | None] = mapped_column("Color", String(15))
    list_price: Mapped[Decimal] = mapped_column("ListPrice", Numeric(19, 4))
    standard_cost: Mapped[Decimal] = mapped_column("StandardCost", Numeric(19, 4))
    size: Mapped[str | None] = mapped_column("Size", String(5))
    product_category_id: Mapped[int | None] = mapped_column("ProductCategoryID", Integer)
    sell_start_date: Mapped[datetime] = mapped_column("SellStartDate", DateTime)
    modified_date: Mapped[datetime] = mapped_column(
        "ModifiedDate", DateTime, server_default=func.getdate()
    )


with Session(engine) as session:
    # Create
    product = Product(
        name="Classic Road Bike",
        product_number="BK-C001",
        color="Red",
        list_price=Decimal("1299.99"),
        standard_cost=Decimal("749.99"),
        sell_start_date=datetime(2026, 1, 1),
        product_category_id=6,
    )
    session.add(product)
    session.commit()
    new_id = product.product_id
    print(f"Inserted ProductID: {new_id}")

    # Read
    product = session.get(Product, new_id)
    print(f"Read: {product.name} costs ${product.list_price}")

    # Update
    product.list_price = Decimal("1349.99")
    session.commit()
    print(f"Updated price to ${product.list_price}")

    # Delete
    session.delete(product)
    session.commit()
    print(f"Deleted ProductID: {new_id}")

Запустите скрипт:

python crud.py

Вы видите выход, похожий на следующий:

Inserted ProductID: 1019
Read: Classic Road Bike costs $1299.9900
Updated price to $1349.9900
Deleted ProductID: 1019

Скрипт удаляет строку, которую он создаёт, поэтому при повторном запуске не нарушает ограничения уникальности для Name и ProductNumber. Каждый запуск добавляет новую строку, поэтому ProductID увеличивается при каждом запуске.

Основные запросы

SQLAlchemy Core предоставляет низкоуровневый API для SQL-выражений. Вы можете использовать Core с тем же движком и определениями таблиц, включая классы, отображаемые в ORM.

from sqlalchemy import text

with engine.connect() as conn:
    result = conn.execute(text("SELECT @@VERSION"))
    print(result.scalar())

Используйте конструкции на уровне таблиц для генерации SQL, безопасной с точки зрения типов:

from sqlalchemy import insert, select, update, delete

with engine.connect() as conn:
    # Insert
    conn.execute(
        insert(Product).values(
            name="Touring Bike",
            product_number="BK-T002",
            list_price=Decimal("999.99"),
            standard_cost=Decimal("575.00"),
            sell_start_date=datetime(2026, 1, 1)
        )
    )
    conn.commit()

    # Select
    stmt = select(
        Product.name.label("name"),
        Product.list_price.label("list_price"),
    ).where(Product.list_price > 100)
    for row in conn.execute(stmt):
        print(row.name, row.list_price)

    # Delete the inserted row so this example can run again
    conn.execute(delete(Product).where(Product.product_number == "BK-T002"))
    conn.commit()

Замечание

Когда вы выбираете отдельные сопоставленные столбцы, имя которых в базе данных отличается от имени атрибута (например, Product.name сопоставляется со столбцом Name), строки Core используют имя столбца базы данных в качестве ключа. Добавьте .label("name") для доступа к значению как row.name вместо row.Name.

Пулинг соединений

SQLAlchemy по умолчанию управляет пулом соединений. Настройте настройки пула под вашу нагрузку:

engine = create_engine(
    "mssql+mssqlpython://dbuser:<password>@localhost/<database>",
    pool_size=10,
    max_overflow=20,
    pool_timeout=30,
    pool_recycle=3600,
)
Parameter Описание
pool_size Количество подключений, которые нужно держать открытыми (по умолчанию: 5).
max_overflow Разрешённые соединения дальше pool_size (по умолчанию: 10).
pool_timeout Секунды на ожидание соединения, прежде чем появится ошибка (по умолчанию: 30).
pool_recycle Через несколько секунд соединение перерабатывается (по умолчанию: -1, отключен). Установите это значение, если ваша база данных закрывает холостые соединения.

Использование с веб-фреймворками

SQLAlchemy обычно используется в качестве уровня базы данных для Flask и FastAPI. Диалект mssql-python работает с любым фреймворком, поддерживающим SQLAlchemy.

Следующие фрагменты показывают рекомендуемый шаблон сессии на запрос для каждого фреймворка. Это иллюстративные фрагменты, которые основаны на модели engineиProduct, описанной в предыдущих разделах, а не полноценные приложения. Для полностью готовых к запуску приложений см. статьи об интеграции FastAPI и об интеграции Flask.

Пример FastAPI

Используйте зависимость генератора, чтобы предоставить сессию для каждого запроса:

from fastapi import Depends, FastAPI, HTTPException
from sqlalchemy.orm import Session, sessionmaker
from sqlalchemy import create_engine

engine = create_engine("mssql+mssqlpython://dbuser:<password>@<server>/<database>")
SessionLocal = sessionmaker(bind=engine)

app = FastAPI()


def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()


@app.get("/products/{product_id}")
def read_product(product_id: int, db: Session = Depends(get_db)):
    product = db.get(Product, product_id)
    if not product:
        raise HTTPException(status_code=404, detail="Product not found")
    return {"name": product.name, "price": float(product.list_price)}

Пример Flask

Используйте менеджер контекста, чтобы ограничить сессию рамками запроса:

from flask import Flask, jsonify
from sqlalchemy.orm import Session, sessionmaker
from sqlalchemy import create_engine

engine = create_engine("mssql+mssqlpython://dbuser:<password>@<server>/<database>")
SessionLocal = sessionmaker(bind=engine)

app = Flask(__name__)


@app.route("/products/<int:product_id>")
def read_product(product_id):
    with SessionLocal() as session:
        product = session.get(Product, product_id)
        if not product:
            return jsonify({"error": "Not found"}), 404
        return jsonify({"name": product.name, "price": float(product.list_price)})

Алембийские миграции

Alembic занимается миграцией схем для проектов SQLAlchemy и работает с диалектом mssql-python. Функция автогенерации Alembic сравнивает ваши модели с живой базой данных, поэтому несколько дополнительных шагов не позволяют ей предлагать изменения в таблицах, которыми вы не управляете.

Настройте Alembic

Установите Alembic и инициализуйте каталог миграций:

pip install alembic
alembic init migrations

В alembic.ini установите URL-адрес подключения:

sqlalchemy.url = mssql+mssqlpython://dbuser:<password>@localhost/<database>

Направьте Alembic на свои модели

Автогенерация требует метаданных ваших моделей. Поместите модели, которыми управляет Alembic, в импортируемый модуль, например models.py. Поскольку автогенерация предлагает убрать любой столбц, который модель опускает, определить модель, которая полностью владеет своей таблицей, а не повторяет упрощённую Product модель, описанную ранее в этой статье:

# models.py
from datetime import datetime

from sqlalchemy import Identity, String, Integer, DateTime, func
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column


class Base(DeclarativeBase):
    pass


class ProductReview(Base):
    __tablename__ = "ProductReview"
    __table_args__ = {"schema": "SalesLT"}

    review_id: Mapped[int] = mapped_column("ReviewID", Integer, Identity(), primary_key=True)
    product_id: Mapped[int] = mapped_column("ProductID", Integer)
    reviewer_name: Mapped[str] = mapped_column("ReviewerName", String(50))
    rating: Mapped[int] = mapped_column("Rating", Integer)
    comments: Mapped[str | None] = mapped_column("Comments", String(500))
    modified_date: Mapped[datetime] = mapped_column("ModifiedDate", DateTime, server_default=func.getdate())

Предостережение

По умолчанию автогенерация считает каждую таблицу в базе данных, которая не входит target_metadata , удаленной и эмитирует drop_table для неё. При работе с существующей базой данных, такой как AdventureWorksLT, это действие может удалить десятки таблиц. Добавьте include_name фильтр, чтобы Alembic управлял только таблицами, которые определяют ваши модели, и всегда проверяйте сгенерированный скрипт перед его применением.

В migrations/env.py замените target_metadata = None на следующий код. Он импортирует ваши модели и ограничивает автогенерацию только схемами и таблицами, которые они определяют:

from models import Base

target_metadata = Base.metadata

# Limit autogenerate to the tables your models define.
managed_schemas = {table.schema for table in target_metadata.tables.values()}
managed_tables = {table.name for table in target_metadata.tables.values()}


def include_name(name, type_, parent_names):
    if type_ == "schema":
        return name in managed_schemas
    if type_ == "table":
        return name in managed_tables
    return True

Передайте include_name и include_schemas=True в context.configure как в run_migrations_offline, так и в run_migrations_online. Настройка include_schemas=True позволяет Alembic видеть таблицы в нестандартных схемах, таких как SalesLT:

context.configure(
    connection=connection,
    target_metadata=target_metadata,
    include_name=include_name,
    include_schemas=True,
)

Сгенерируйте и примените миграцию

Сгенерируйте миграцию из ваших моделей:

alembic revision --autogenerate -m "add product review table"

Alembic обнаруживает новую таблицу и пишет скрипт миграции:

INFO  [alembic.autogenerate.compare.tables] Detected added table 'SalesLT.ProductReview'
Generating .../versions/xxxx_add_product_review_table.py ... done

Сгенерированный upgrade() создаёт таблицу и downgrade() сбрасывает её:

def upgrade() -> None:
    op.create_table(
        "ProductReview",
        sa.Column("ReviewID", sa.Integer(), sa.Identity(always=False), nullable=False),
        sa.Column("ProductID", sa.Integer(), nullable=False),
        sa.Column("ReviewerName", sa.String(length=50), nullable=False),
        sa.Column("Rating", sa.Integer(), nullable=False),
        sa.Column("Comments", sa.String(length=500), nullable=True),
        sa.Column("ModifiedDate", sa.DateTime(), server_default=sa.text("getdate()"), nullable=False),
        sa.PrimaryKeyConstraint("ReviewID"),
        schema="SalesLT",
    )


def downgrade() -> None:
    op.drop_table("ProductReview", schema="SalesLT")

Просмотрите скрипт, а затем примените все незавершённые миграции:

alembic upgrade head

Отличия от диалекта pyodbc

Если вы переходите с mssql+pyodbc, диалект mssql-python похож, потому что оба драйвера основаны на одном и том же фреймворке ODBC. Основные различия:

Тема mssql+pyodbc mssql+mssqlpython
Установка драйверов ODBC Требуется отдельный драйвер ODBC (например, ODBC Driver 18 для Microsoft SQL). Драйвер в комплекте. Отдельный драйвер ODBC не требуется.
URL-адрес соединения mssql+pyodbc://user:pass@host/db?driver=ODBC+Driver+18+for+SQL+Server mssql+mssqlpython://user:pass@host/db
fast_executemany Поддерживается через create_engine(..., fast_executemany=True). Неприменимо. Драйвер самостоятельно обеспечивает пакетную обработку.
Availability Стабильный, включённый в SQLAlchemy начиная с версии 1.x. Предварительный выпуск (SQLAlchemy 2.1.0b2+).

Известные ограничения

Диалект mssql-python для SQLAlchemy находится в предварительном релизе. Перед использованием в производстве понимайте следующие последствия:

  • Изменения в API: Сигнатуры методов, типы исключений и поведение могут измениться до финального стабильного релиза. Всегда привязывайте свою версию SQLAlchemy к конкретной сборке до релиза (например) sqlalchemy==2.1.0b2и тщательно тестируйте обновления.

  • Ограниченное тестирование: У диалекта меньше тестирования в сообществах, чем у стабильного mssql+pyodbc . Вы можете столкнуться с крайними случаями или отсутствующими функциями.

  • Пробелы в функциях: некоторые продвинутые функции ORM или Core могут не работать. Обратитесь к документации по диалекту SQLAlchemy MSSQL и протестируйте свои сценарии использования перед тем, как браться за проект.

  • Нет гарантии поддержки: Microsoft и SQLAlchemy обеспечивают поддержку с максимальными усилиями, но проблемы могут не быть решены до стабильного релиза.

Когда использовать пре-релиз:

  • среды для разработки и тестирования;
  • Проекты по демонстрации концепции
  • Переход с mssql+pyodbc, если вы хотите избежать зависимости от внешнего драйвера ODBC
  • Проекты, где можно реагировать на изменения API и проводить регрессионное тестирование

Когда НЕ стоит использовать пре-релиз:

  • Производственные системы с строгими требованиями к стабильности
  • Многолетние устаревшие приложения, где обновления зависимостей встречаются редко
  • Критически важные бизнес-нагрузки до тех пор, пока SQLAlchemy 2.1 не получит статус общедоступной версии

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

Troubleshooting

"Нет модуля с названием 'sqlalchemy.dialects.mssql.mssqlpython'"

Эта ошибка означает, что ваша установленная версия SQLAlchemy не включает диалект mssql-python. Убедитесь, что у вас есть версия 2.1.0b2 или более поздняя:

pip install "sqlalchemy>=2.1.0b2"

Сбои подключения

Если create_engine удаётся, но запросы не работают, проверьте, что параметры соединения работают напрямую с mssql-python:

import mssql_python

conn = mssql_python.connect(
    "Server=localhost;Database=<database>;UID=dbuser;PWD=<password>;Encrypt=yes"
)
cursor = conn.cursor()
cursor.execute("SELECT 1")
print(cursor.fetchone())
conn.close()

Если прямое соединение работает, а SQLAlchemy — нет, проверьте проблемы с кодированием URL в специальных символах в вашем пароле или имени сервера.

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

  • Last updated on