使用 MSSQL-Python 搭配 SQLAlchemy

SQLAlchemy 是最廣泛使用的 Python ORM 與資料庫工具包。 SQLAlchemy 2.1 內建了適用於 mssql-python 驅動程式的方言，可讓您將 SQLAlchemy ORM 和 Core 與 Microsoft SQL 及 Azure SQL Database 搭配使用。

先決條件

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 定義為 mssql-python>=1.9.0。同時也支援現有的分離套件形式：

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 認證，請在連線網址中包含使用者名稱和密碼：

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"
)

Note

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

使用 SQLAlCHIS 的宣告式映射來定義對應到 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}")

Note

在 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 替換成您自己的（參見連線網址）

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

腳本會刪除它建立的那一列，所以當你再次執行時，它不會遇到和 NameProductNumber 上的獨特限制。每次執行都會插入一個新列，因此 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()

Note

當你選擇資料庫名稱與屬性名稱不同的個別映射欄位（例如， Product.name 映射到欄位 Name ）時，核心列會依資料庫欄位名稱來鍵化。加入 .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，停用）。如果你的資料庫關閉閒置連線，請設定這個值。

搭配網頁框架使用

SQLAlchemy 常被用作 Flask 和 FastAPI 的資料庫層。 mssql-python 方言適用於任何支援 SQLAlcMMY 的框架。

以下摘要展示了每個框架建議的每次請求會話模式。它們是範例片段，假設了前面部分的和 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中，設定連線網址：

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 ，因為 SQLAlchemy 2.1 已不再預設安裝 greenlet 。

Troubleshooting

「沒有名為 'sqlalchemy.dialects.mssql.mssqlpython' 的模組」

這個錯誤表示你已安裝的 SQLAlchemy 版本不包含 mssql-python 方言。安裝 SQLAlchemy 2.1 或更新版本，並安裝其 mssql-python 相依套件：

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 編碼問題。

意見反應

此頁面對您有幫助嗎？

Last updated on 2026-09-25

使用 MSSQL-Python 搭配 SQLAlchemy

先決條件

安裝 SQLAlchemy 和 mssql-python

連線網址

SQL 驗證

Microsoft Entra 認證

以程式化方式建置 URL

定義 ORM 模型

CRUD作業

創建會話

插入數據列

查詢列

更新資料列

刪除列

完整範例

核心查詢

連線池化

搭配網頁框架使用

FastAPI 範例

Flask 範例

Alembic 遷移

設立Alembic

讓 Alembic 指向你的模型

產生並套用遷移

與 pyodbc 方言的差異

升級時的考量

Troubleshooting

「沒有名為 'sqlalchemy.dialects.mssql.mssqlpython' 的模組」

連線失敗

相關內容

意見反應

其他資源