使用 mssql-python 配合 Flask

Flask 是一个轻量级的 Python Web 框架，能让你完全控制应用结构。结合mssql-python，你可以以最小的开销构建由Microsoft SQL和Azure SQL 数据库支持的Web应用和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)

注释

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

注释

在 PowerShell 中，curl 是 Invoke-WebRequest 的别名。这里简单的 GET 命令运行正常，但响应会以对象形式返回，而不是打印出来的 JSON。使用 -X、-H 或 -d 等 curl 标志的命令（例如后文的 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 中，和 NameProductNumber 都有唯一的约束，所以测试每次都会为每个变量生成一个唯一的值。如果改为硬编码这些值，除非先删除该行，否则测试在第二次运行时会因冲突而失败。

运行测试

把测试保存在 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