使用 mssql-python 搭配 Flask

Flask 是一個輕量級的 Python 網頁框架，讓你能完全掌控應用程式結構。結合 mssql-python，你可以以最小的開銷建立由 Microsoft SQL 和 Azure SQL Database 支援的網頁應用程式和 REST API。

先決條件

Python 3.10 或更新版本。
mssql-python 與 flask 套件。使用 pip install flask mssql-python 安裝兩者。

安裝一次性作業系統特定先決條件。 Windows 使用者可以跳過此步驟。完整平台細節請參見安裝 mssql-python。

apk add libtool krb5-libs krb5-dev

apt-get install -y libltdl7 libkrb5-3 libgssapi-krb5-2

dnf install -y libtool-ltdl krb5-libs

zypper install -y libltdl7 libkrb5-3 libgssapi-krb5-2

zypper install -y libltdl7

brew install openssl

建立 SQL 資料庫

在以下平台建立或連接 SQL 資料庫：

使用 Azure 入口網站在幾分鐘內建立 SQL 資料庫

本文中的範例使用 AdventureWorksLT 範例資料庫，尤其是 SalesLT.Product 資料表。如果你還沒安裝 AdventureWorksLT，請參考 AdventureWorks 範例資料庫。

專案設定

安裝依賴項

使用 pip 安裝所需套件：

pip install flask mssql-python

專案結構

用獨立模組來組織你的專案，分別負責設定、連線管理、路由和測試：

my_app/
├── app.py            # Flask app and routes
├── config.py         # database settings
├── database.py       # connection lifecycle
├── test_app.py       # pytest tests
└── blueprints/       # optional: routes grouped into modules
    ├── __init__.py
    └── products.py

資料庫連線管理

Flask 沒有內建資料庫層，所以你直接管理連線。本節的模式會在 Flask g 物件上儲存每個請求一個連線，並在請求結束時自動關閉該連線。

建立 config.py

將資料庫設定集中管理於設定類別中。環境變數讓你可以在不改程式碼的情況下覆寫預設值。

# config.py
import os

class Config:
    """Application configuration."""
    DATABASE_SERVER = os.getenv("DB_SERVER", "<server>.database.windows.net")
    DATABASE_NAME = os.getenv("DB_NAME", "<database>")
    POOL_SIZE = int(os.getenv("DB_POOL_SIZE", "10"))

建立 database.py

該 database.py 模組負責管理連線生命週期。 Flask 的 g 物件是每個請求專屬的命名空間，因此將連線儲存在其中，可確保每個請求都有自己的連線，並在請求結束時加以清理。

get_connection_string() 函式會根據應用程式設定建立連線字串。get_db() 函式會在第一次呼叫時建立連線，並在該要求的其餘期間重複使用該連線。該 close_db() 函式會在每次請求結束時自動執行，若發生例外則回滾交易，否則則提交。該 init_app() 函式會透過 Flask 應用程式登錄此拆解行為。

# database.py
import mssql_python
from flask import g, current_app

def get_connection_string() -> str:
    """Build connection string from Flask app config."""
    cfg = current_app.config
    return (
        f"Server={cfg['DATABASE_SERVER']};"
        f"Database={cfg['DATABASE_NAME']};"
        "Authentication=ActiveDirectoryDefault;"
        "Encrypt=yes"
    )

def get_db():
    """Get a database cursor for the current request.

    The connection is stored on Flask's g object so it persists
    for the duration of the request and is reused across calls.
    """
    if "db_conn" not in g:
        g.db_conn = mssql_python.connect(get_connection_string())
        g.db_cursor = g.db_conn.cursor()
    return g.db_cursor

def close_db(exception=None):
    """Close the database connection at the end of the request."""
    cursor = g.pop("db_cursor", None)
    conn = g.pop("db_conn", None)

    if cursor is not None:
        cursor.close()
    if conn is not None:
        if exception:
            conn.rollback()
        else:
            conn.commit()
        conn.close()

def init_app(app):
    """Register database teardown with the Flask app."""
    app.teardown_appcontext(close_db)

Note

ActiveDirectoryDefault 使用 DefaultAzureCredential，其會依序嘗試多個憑證提供者。第一次連線可能會比較慢，因為 SDK 會一直走鏈條直到找到可用的供應商。在生產環境中，如果您知道您的環境使用哪一種憑證類型，請直接指定該憑證類型（例如，針對受控識別可指定 ActiveDirectoryMSI），以避免逐一嘗試整個憑證鏈。如需詳細資訊，請參閱 Microsoft Entra 驗證。

Flask 應用程式

以下範例展示了完整的 Flask 應用程式，包含列出、檢索、建立、更新及刪除產品的路徑。

建立 app.py

應用程式模組建立 Flask 應用程式，載入設定，並登錄資料庫拆解。每個路由函式會呼叫 get_db() 取得游標，並以參數化 SQL （使用 %(name)s 佔位符與值字典）執行查詢，並回傳 JSON 回應。

# app.py
from flask import Flask, jsonify, request, abort
from config import Config
from database import init_app, get_db

app = Flask(__name__)
app.config.from_object(Config)
init_app(app)

@app.route("/")
def index():
    return jsonify({"message": "Product API", "docs": "/products"})

@app.route("/products")
def list_products():
    """List products with pagination."""
    page = request.args.get("page", 1, type=int)
    page_size = request.args.get("page_size", 10, type=int)
    skip = (page - 1) * page_size

    cursor = get_db()

    cursor.execute("SELECT COUNT(*) FROM SalesLT.Product")
    total = cursor.fetchval()

    cursor.execute("""
        SELECT ProductID, Name, ProductNumber, ListPrice, Color, ProductCategoryID
        FROM SalesLT.Product
        ORDER BY ProductID
        OFFSET %(skip)s ROWS
        FETCH NEXT %(limit)s ROWS ONLY
    """, {"skip": skip, "limit": page_size})

    items = [{
        "id": row.ProductID,
        "name": row.Name,
        "product_number": row.ProductNumber,
        "price": float(row.ListPrice),
        "color": row.Color,
        "category_id": row.ProductCategoryID
    } for row in cursor.fetchall()]

    return jsonify({
        "items": items,
        "total": total,
        "page": page,
        "page_size": page_size,
        "pages": (total + page_size - 1) // page_size
    })

@app.route("/products/<int:product_id>")
def get_product(product_id):
    """Get a single product by ID."""
    cursor = get_db()
    cursor.execute("""
        SELECT ProductID, Name, ProductNumber, ListPrice, Color, ProductCategoryID
        FROM SalesLT.Product
        WHERE ProductID = %(id)s
    """, {"id": product_id})

    row = cursor.fetchone()
    if not row:
        abort(404)

    return jsonify({
        "id": row.ProductID,
        "name": row.Name,
        "product_number": row.ProductNumber,
        "price": float(row.ListPrice),
        "color": row.Color,
        "category_id": row.ProductCategoryID
    })

@app.route("/products", methods=["POST"])
def create_product():
    """Create a new product."""
    data = request.get_json()
    if not data:
        abort(400)

    cursor = get_db()

    # OUTPUT INSERTED returns the new row's columns in the same statement,
    # so you don't need a separate SELECT to get the generated ID and defaults.
    # ProductNumber is required and unique. StandardCost and SellStartDate are
    # also NOT NULL in SalesLT.Product, so supply values for them.
    cursor.execute("""
        INSERT INTO SalesLT.Product
            (Name, ProductNumber, ListPrice, Color, Size, ProductCategoryID, StandardCost, SellStartDate)
        OUTPUT INSERTED.ProductID, INSERTED.Name, INSERTED.ProductNumber, INSERTED.ListPrice,
               INSERTED.Color, INSERTED.ProductCategoryID
        VALUES (%(name)s, %(product_number)s, %(price)s, %(color)s, %(size)s, %(category_id)s, 0, GETDATE())
    """, {
        "name": data["name"],
        "product_number": data["product_number"],
        "price": data["price"],
        "color": data.get("color"),
        "size": data.get("size"),
        "category_id": data["category_id"]
    })

    row = cursor.fetchone()
    return jsonify({
        "id": row.ProductID,
        "name": row.Name,
        "product_number": row.ProductNumber,
        "price": float(row.ListPrice),
        "color": row.Color,
        "category_id": row.ProductCategoryID
    }), 201

@app.route("/products/<int:product_id>", methods=["PUT"])
def update_product(product_id):
    """Update an existing product."""
    data = request.get_json()
    if not data:
        abort(400)

    cursor = get_db()

    updates = []
    params = {"id": product_id}

    for field in ("name", "product_number", "price", "color", "category_id"):
        if field in data:
            col = {"name": "Name", "product_number": "ProductNumber",
                   "price": "ListPrice", "color": "Color",
                   "category_id": "ProductCategoryID"}[field]
            updates.append(f"{col} = %({field})s")
            params[field] = data[field]

    if not updates:
        abort(400)

    cursor.execute(f"""
        UPDATE SalesLT.Product SET {', '.join(updates)}
        OUTPUT INSERTED.ProductID, INSERTED.Name, INSERTED.ProductNumber, INSERTED.ListPrice,
               INSERTED.Color, INSERTED.ProductCategoryID
        WHERE ProductID = %(id)s
    """, params)

    row = cursor.fetchone()
    if not row:
        abort(404)

    return jsonify({
        "id": row.ProductID,
        "name": row.Name,
        "product_number": row.ProductNumber,
        "price": float(row.ListPrice),
        "color": row.Color,
        "category_id": row.ProductCategoryID
    })

@app.route("/products/<int:product_id>", methods=["DELETE"])
def delete_product(product_id):
    """Delete a product."""
    cursor = get_db()
    cursor.execute("DELETE FROM SalesLT.Product WHERE ProductID = %(id)s", {"id": product_id})
    if cursor.rowcount == 0:
        abort(404)
    return "", 204

@app.route("/health")
def health_check():
    """Check database connectivity."""
    try:
        cursor = get_db()
        cursor.execute("SELECT 1")
        return jsonify({"status": "healthy", "database": "connected"})
    except Exception as e:
        return jsonify({"status": "unhealthy", "error": str(e)}), 503

執行應用程式

啟動開發伺服器：

flask --app app run --debug --port 5000

伺服器會在 http://localhost:5000 上監聽。打開第二個終端機，並透過以下方式 curl 呼叫端點，以確認應用程式是否與你的資料庫通訊：

# Check database connectivity
curl http://localhost:5000/health

# List the first page of products
curl "http://localhost:5000/products?page_size=5"

# Get a single product by ID
curl http://localhost:5000/products/680

Note

在 PowerShell 中， curl 是的別名 Invoke-WebRequest。這裡簡單的 GET 指令執行得很好，但回應會以物件形式回傳，而不是列印出來的 JSON。使用 curl 旗標如 -X、 -H或 -d （如後面範例 POST ）的指令，並不會照寫法運作。在 Windows 上，請依curl.exe照顯示的指令執行，或使用 PowerShell Invoke-RestMethod （例如 Invoke-RestMethod http://localhost:5000/health），它也會幫你解析 JSON 回應。

每個端點都會回傳 JSON。你也可以在瀏覽器中開啟 http://localhost:5000/products 查看分頁清單。

連線池化

若沒有連線池，每個請求都會開啟或關閉 Microsoft SQL 的 TCP 連線，這會增加延遲。連線池會讓一組閒置連線隨時待命，方便重複使用。若要啟用連線集區，請在模組層級呼叫一次 mssql_python.pooling()。啟用集區化時，在 close_db 拆解期間，conn.close() 會將連線返回集區，而不是關閉該連線。

啟用連線池

在開啟任何連線前，於模組層級呼叫 mssql_python.pooling() 以啟用池化：

# database.py with connection pooling
import mssql_python
from flask import g, current_app

# Configure pool at module level
mssql_python.pooling(max_size=20, idle_timeout=300)

def get_db():
    """Get a database cursor with connection pooling."""
    if "db_conn" not in g:
        g.db_conn = mssql_python.connect(get_connection_string())
        g.db_cursor = g.db_conn.cursor()
    return g.db_cursor

錯誤處理

Flask 允許你為特定例外類型註冊處理器。攔截 mssql_python.DatabaseError 和 mssql_python.IntegrityError 可讓您傳回結構化的 JSON 錯誤回應，而不是預設的 HTML 錯誤頁面。

註冊錯誤處理常式

將這些處理常式加入現有的 app.py 中，並放在 app = Flask(__name__) 這一行之後。由於處理常式會參照 app 物件，因此它們必須在建立應用程式之後才宣告。 app.py需要將 import mssql_python 放在頂端。處理器會回傳結構化的 JSON 回應，而非預設的 HTML 錯誤頁面：

# app.py
import mssql_python

@app.errorhandler(mssql_python.DatabaseError)
def handle_database_error(error):
    """Handle database errors."""
    return jsonify({"error": "Database error occurred"}), 500

@app.errorhandler(mssql_python.IntegrityError)
def handle_integrity_error(error):
    """Handle integrity constraint violations."""
    error_msg = str(error)
    if "UNIQUE" in error_msg:
        return jsonify({"error": "Resource already exists"}), 409
    if "FOREIGN KEY" in error_msg:
        return jsonify({"error": "Referenced resource not found"}), 400
    return jsonify({"error": "Data integrity error"}), 400

@app.errorhandler(404)
def not_found(error):
    return jsonify({"error": "Resource not found"}), 404

@app.errorhandler(400)
def bad_request(error):
    return jsonify({"error": "Bad request"}), 400

Blueprints

隨著應用程式成長，將所有路由放在單一檔案中會變得困難。 Flask Blueprints 允許你將相關路由分組成獨立模組，並註冊在應用程式中。

用藍圖組織路線

建立一個產品路由藍圖模組，匯入 get_db 並定義共用 URL 前綴下的端點：

# blueprints/products.py
from flask import Blueprint, jsonify, request, abort
from database import get_db

products_bp = Blueprint("products", __name__, url_prefix="/api/products")

@products_bp.route("/")
def list_products():
    """List all products."""
    cursor = get_db()
    cursor.execute("""
        SELECT ProductID, Name, ListPrice, Color, ProductCategoryID
        FROM SalesLT.Product ORDER BY ProductID
    """)
    return jsonify([{
        "id": row.ProductID,
        "name": row.Name,
        "price": float(row.ListPrice),
        "color": row.Color,
        "category_id": row.ProductCategoryID
    } for row in cursor.fetchall()])

@products_bp.route("/<int:product_id>")
def get_product(product_id):
    """Get a product by ID."""
    cursor = get_db()
    cursor.execute(
        "SELECT ProductID, Name, ListPrice, Color FROM SalesLT.Product WHERE ProductID = %(id)s",
        {"id": product_id}
    )
    row = cursor.fetchone()
    if not row:
        abort(404)
    return jsonify({"id": row.ProductID, "name": row.Name, "price": float(row.ListPrice), "color": row.Color})

註冊藍圖

將藍圖存為 blueprints/products.py，並新增一個空blueprints/__init__.py檔案，讓 Python 將該資料夾視為套件。接著在 app.py，將藍圖與其他匯入一起匯入，並在該行 app = Flask(__name__) 後註冊：

# app.py
from blueprints.products import products_bp

app.register_blueprint(products_bp)

由於藍圖將 url_prefix="/api/products" 設為前綴，其路由會在該前綴下提供。例如，列表路由可在 http://localhost:5000/api/products/ 使用，與直接在 app.py 中定義的 /products 路由分開。

Testing

Flask 提供一個測試客戶端，可以在不啟動真實 HTTP 伺服器的情況下，向你的應用程式發送請求。用 pytest 夾具建立客戶端，並在測試中重複使用。

使用 pytest 進行測試設定

建立一個 pytest 夾具，提供測試客戶端並撰寫測試以驗證路由行為：

# test_app.py
import uuid

import pytest
from app import app

@pytest.fixture
def client():
    app.config["TESTING"] = True
    with app.test_client() as client:
        yield client

def test_health_check(client):
    response = client.get("/health")
    assert response.status_code == 200
    data = response.get_json()
    assert data["status"] == "healthy"

def test_list_products(client):
    response = client.get("/products")
    assert response.status_code == 200
    data = response.get_json()
    assert "items" in data
    assert "total" in data

def test_create_product(client):
    suffix = uuid.uuid4().hex[:8]
    name = f"Test Product {suffix}"
    response = client.post("/products", json={
        "name": name,
        "product_number": f"TEST-{suffix}",
        "price": 19.99,
        "category_id": 18
    })
    assert response.status_code == 201
    data = response.get_json()
    assert data["name"] == name

def test_get_product_not_found(client):
    response = client.get("/products/99999")
    assert response.status_code == 404

這些測試是針對你的即時資料庫執行，而非模擬資料庫，因此test_create_product會插入一個實數列。SalesLT.Product 在 AdventureWorksLT 中，Name 和 ProductNumber 都具有唯一約束，因此每次執行測試時，都會為兩者各自產生唯一值。如果你改為將那些值硬編碼，除非先刪除那一列，否則測試在第二次執行時會因衝突而失敗。

執行測試

把測試資料存到 test_app.py 專案資料夾裡。啟用虛擬環境後，從那個資料夾安裝 pytest 並執行。在與 flask 和 mssql-python 相同的虛擬環境中安裝並執行 pytest，可確保測試會匯入你的應用程式所使用的套件。 pytest 自動 test_app.py 發現並回報結果：

pip install pytest
pytest

pytest 自動發現 test_app.py 並回報結果：

==================== test session starts ====================
collected 4 items

test_app.py ....                                       [100%]

===================== 4 passed in 3.21s =====================

意見反應

此頁面對您有幫助嗎？

Last updated on 2026-07-22