将 mssql-python 与 SQLAlchemy 结合使用

SQLAlchemy 是最广泛使用的 Python ORM 和数据库工具包。 从 SQLAlchemy 2.1.0b2 开始，针对 mssql-python 驱动的内置方言让你可以将 SQLAlchemy ORM 和 Core 用于 Microsoft SQL 和 Azure SQL 数据库。

Important

mssql-python方言是在 SQLAlchemy 2.1.0b2（2026年4月16日发布）中添加的。 SQLAlchemy 2.1 目前是 预发布 系列， 不推荐用于生产环境。 在从 SQLAlchemy 2.0升级之前，请理解：

• API 可能会在最终稳定版（2.1 GA）之前发生变化
• 部署前要彻底测试你的工作量
• 在 2.1 正式发布之前，生产系统请使用稳定的 SQLAlchemy 2.0.x。
• 将依赖项固定为特定版本（例如 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

连接网址

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 模型

使用 SQLAlcMMY 的声明式映射来定义映射到 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 提供了一个低级别的 SQL 表达式 API。 你可以在 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,
)

参数	描述
pool_size	保持打开的连接数（默认值：5）。
max_overflow	超过 pool_size 后允许的连接数（默认值：10）。
pool_timeout	报错前等待连接的秒数（默认值：30）。
pool_recycle	几秒钟后连接会被回收（默认：-1，禁用）。 如果你的数据库关闭了空闲连接，请设置这个值。

与 Web 框架配合使用

SQLAlchemy 通常作为 Flask 和 FastAPI 的数据库层。 mssql-python方言适用于任何支持 SQLAlchemy的框架。

以下摘要展示了每个框架推荐的每次请求会话模式。 它们是示例片段，假设了之前部分的建模engineProduct，而不是完整的应用。 关于完整且可运行的应用程序，请参见 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 迁移

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

在 run_migrations_offline 和 run_migrations_online 中，都将 include_name 和 include_schemas=True 传递给 context.configure。 该 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 驱动（例如，Microsoft SQL 的 ODBC 驱动 18）。	已附带驱动程序。 不需要单独的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	稳定，自1.x版本起就包含在 SQLAlchemy 中。	预发布版（SQLAlchemy 2.1.0b2+）。

已知限制

SQLAlchemy的mssql-python方言还处于预发布阶段。 在生产环境中使用之前，请理解以下含义：

• API 变更：方法签名、异常类型和行为可能会在最终稳定版发布前发生变化。 务必将你的 SQLAlchemy 版本固定为特定的预发布构建版本（例如 sqlalchemy==2.1.0b2），并对升级进行充分测试。

• 有限测试：该方言的社区测试比稳定 mssql+pyodbc 方言少。 你可能会遇到边缘场景或功能缺失。

• 功能缺口：一些高级 ORM 或核心功能可能无法使用。 在投入项目前，请参考 SQLAlchemy MSSQL 方言文档 并测试你的用例。

• 无支持保证：Microsoft 和 SQLAlchemy 提供尽力支持，但问题可能在稳定版发布前无法解决。

何时使用预发布：

• 开发和测试环境
• 概念验证项目
• 如果你想避免对外部 ODBC 驱动程序的依赖，可从 mssql+pyodbc 迁移
• 这些项目可以响应API变更并进行回归测试

何时不应使用预发布：

• 具有严格稳定性要求的生产系统
• 依赖项很少更新的多年遗留应用程序
• 关键业务工作负载，直到 SQLAlchemy 2.1 达到稳定 GA

有关最新的预发布方言状态和已知问题，请查看 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 编码问题。

相关内容

• MSSQL-Python 连接字符串
• MSSQL-Python 连接管理
• FastAPI 集成
• Flask 集成
• SQLAlchemy MSSQL 方言文档