将 mssql-python 与 SQLAlchemy 结合使用

SQLAlchemy 是最广泛使用的 Python ORM 和数据库工具包。 SQLAlchemy 2.1 包含用于 mssql-python 驱动程序的内置方言，你可以使用它通过 SQLAlchemy ORM 和 Core 操作 Microsoft SQL 和 Azure SQL 数据库。

先决条件

• Python 3.11 或更高版本。 SQLAlchemy 2.1 停止支持 Python 3.10 及更早版本。
• mssql-python 和 sqlalchemy 软件包（2.1 或更高版本）。

本文中的示例使用了 AdventureWorksLT 示例数据库。 如果你还没有安装AdventureWorksLT，请参见 AdventureWorks示例数据库。

安装 SQLAlchemy 和 mssql-python

使用 SQLAlchemy的 mssql-python 可选依赖来安装这两个包：

pip install "sqlalchemy[mssql-python]>=2.1"

SQLAlchemy 定义了包含 mssql-python>=1.9.0 的 mssql-python可选依赖组。 现有的独立包形式也被支持：

pip install "sqlalchemy>=2.1" mssql-python

验证已安装的版本：

import sqlalchemy
print(sqlalchemy.__version__)  # Should show 2.1.0 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,
)

参数	Description
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

在 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 框架。 主要区别：

Topic	mssql+pyodbc	mssql+mssqlpython
ODBC 驱动安装	需要独立的ODBC驱动（例如，SQL Server的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) 提供支持。	不適用。 驱动程序内部处理批处理性能。
可用 性	稳定，自1.x版本起就包含在 SQLAlchemy 中。	稳定版，包含在 SQLAlchemy 2.1 或更高版本中。

升级注意事项

SQLAlchemy 2.1 包含可能影响应用从早期版本升级的行为变更：

• SQLAlchemy 2.1 需要 Python 3.11 或更高版本。
• 先把 SQLAlchemy 1.x 应用升级到 SQLAlchemy 2.0，再升级到 2.1。
• 将现有的 SQLAlchemy 2.0 应用与《SQLAlchemy 2.1 的新内容》中描述的行为变化进行测试。

本文示例使用 SQLAlcMMY 的同步 API。 使用 SQLAlchemy asyncio 支持的应用程序必须安装额外版本 asyncio ，因为 SQLAlchemy 2.1 已不再默认 greenlet 安装。

故障排除

“没有名为'sqlalchemy.dialects.mssql.mssqlpython'的模块”

这个错误意味着你安装的 SQLAlchemy 版本不包含 mssql-python 方言。 安装带有 mssql-python 依赖的 SQLAlchemy 2.1 或更高版本：

pip install --upgrade "sqlalchemy[mssql-python]>=2.1"

连接失败

如果 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 管理连接
• 使用 mssql-python 配合 FastAPI
• 使用 mssql-python 配合 Flask
• SQLAlchemy MSSQL 方言文档