FastAPI 是一个现代的 Python 网络框架，用于构建 API。 结合mssql-python，你可以构建由Microsoft SQL和Azure SQL 数据库支持的高性能REST API。
先决条件
- Python 3.10 或更高版本。
- 安装一次性操作系统特定的先决程序。 Windows 用户可以跳过这一步。 完整平台详情请参见 “安装 mssql-python”。
创建 SQL 数据库
在以下平台之一创建或连接SQL数据库：
本文中的示例使用 AdventureWorksLT 示例数据库，具体来说，是 SalesLT.Product 表。 如果你还没有安装AdventureWorksLT，请参见 AdventureWorks示例数据库。
项目设置
创建虚拟环境
创建并激活虚拟环境，使本项目的软件包与其他 Python 安装保持隔离。 这一步还避免了在运行应用或用另一个解释器测试时，将包安装到一个解释器中常见的问题。
py -m venv .venv
.\.venv\Scripts\Activate.ps1
激活该环境后，python、pip 和 pytest 都会解析为同一个解释器。 在已激活的环境中运行本文剩余的命令。
注释
在 Windows on Arm 上，请使用 Arm64 版本的 Python 创建环境，以便 mssql-python 及其依赖项通过预构建的 wheel 包进行安装。 在有多个Python版本的机器上，py -m venv可能会选择与预期不同的版本或架构，激活后请确认python -c "import sys, sysconfig; print(sys.version, sysconfig.get_platform())"。 如果 pip 尝试从源代码构建 cryptography （出现 Rust 和 OpenSSL 工具链错误），请先使用 pip install --only-binary=:all: cryptography 安装基于 wheel 的版本，然后再安装其余内容。
安装依赖项
用 PIP 安装所需的软件包：
pip install fastapi uvicorn mssql-python pydantic
项目结构
用独立模块组织你的项目，分别用于数据库、模式和CRUD操作：
my_api/
├── main.py
├── database.py
├── models.py
├── schemas.py
├── crud.py
└── routers/
└── products.py
数据库连接管理
FastAPI 使用依赖注入向路由处理函数提供数据库连接等资源。 本节中的模式会打开连接、返回游标，并使用 mssql-python 连接上下文管理器在成功时提交、在发生异常时回滚，并关闭连接。
创建 database.py
该get_connection_string()函数从配置值构建ODBC 连接字符串。 FastAPI 对每个请求都会调用一次 Depends()get_db_dependency()，并负责管理其生命周期。
# database.py
import mssql_python
from collections.abc import Generator
# Configuration
DATABASE_CONFIG = {
"server": "<server>.database.windows.net",
"database": "<database>",
}
def get_connection_string() -> str:
"""Build connection string from config."""
return (
f"Server={DATABASE_CONFIG['server']};"
f"Database={DATABASE_CONFIG['database']};"
"Authentication=ActiveDirectoryDefault;"
"Encrypt=yes"
)
def get_db_dependency() -> Generator:
"""FastAPI dependency for database cursor."""
with mssql_python.connect(get_connection_string()) as conn:
with conn.cursor() as cursor:
yield cursor
注释
ActiveDirectoryDefault 使用 DefaultAzureCredential，该程序依次尝试多个凭证提供者。 第一次连接可能很慢，因为SDK会在链路上走动，直到找到可用的提供者。 在生产环境中，如果你知道环境使用的是哪种凭据类型，可以直接指定它（例如，对于托管标识可指定 ActiveDirectoryMSI），以避免遍历凭据链。 有关详细信息，请参阅 Microsoft Entra 身份验证。
Pydantic 模型
Pydantic 模型定义了请求和响应数据的结构及验证规则。 FastAPI 利用这些模型解析接收的 JSON，验证字段约束，并自动生成 OpenAPI 文档。
创建 schemas.py
将模式 Base分为 、 Create、 Update和 响应变体。
Base 模式包含共享字段，用于插入操作的 Create 继承自该模式，而 Update 则使所有字段在部分更新时都变为可选。
# schemas.py
from pydantic import BaseModel, ConfigDict, EmailStr, Field
from typing import Optional
from datetime import datetime
# Product schemas
class ProductBase(BaseModel):
name: str = Field(..., min_length=1, max_length=100)
product_number: str = Field(..., min_length=1, max_length=25)
price: float = Field(..., gt=0)
color: Optional[str] = Field(None, max_length=50)
size: Optional[str] = Field(None, max_length=50)
category_id: Optional[int] = None
class ProductCreate(ProductBase):
pass
class ProductUpdate(BaseModel):
name: Optional[str] = Field(None, min_length=1, max_length=100)
product_number: Optional[str] = Field(None, min_length=1, max_length=25)
price: Optional[float] = Field(None, gt=0)
color: Optional[str] = Field(None, max_length=50)
size: Optional[str] = Field(None, max_length=50)
category_id: Optional[int] = None
class Product(ProductBase):
id: int
model_config = ConfigDict(from_attributes=True)
# Pagination
class PaginatedResponse(BaseModel):
items: list
total: int
page: int
page_size: int
pages: int
CRUD 操作
将数据库查询封装在专用类中，以保持路由处理程序的精简。 每个静态方法都用一个光标（由 FastAPI 注入），并用参数 化查询 （%(name)s 带有值字典的占位符）处理一次操作，以防止 SQL 注入。 这种分离使业务逻辑更容易测试和复用。
创建 crud.py
# crud.py
from typing import Optional, List
from schemas import ProductCreate, ProductUpdate, Product
class ProductCRUD:
"""CRUD operations for products."""
@staticmethod
def get(cursor, product_id: int) -> Optional[dict]:
cursor.execute("""
SELECT ProductID, Name, ProductNumber, ListPrice, Color, Size
FROM SalesLT.Product
WHERE ProductID = %(id)s
""", {"id": product_id})
row = cursor.fetchone()
if row:
return {
"id": row.ProductID,
"name": row.Name,
"product_number": row.ProductNumber,
"price": float(row.ListPrice),
"color": row.Color,
"size": row.Size
}
return None
@staticmethod
def get_all(cursor, skip: int = 0, limit: int = 100) -> List[dict]:
cursor.execute("""
SELECT ProductID, Name, ProductNumber, ListPrice, Color, Size
FROM SalesLT.Product
ORDER BY ProductID
OFFSET %(skip)s ROWS
FETCH NEXT %(limit)s ROWS ONLY
""", {"skip": skip, "limit": limit})
return [{
"id": row.ProductID,
"name": row.Name,
"product_number": row.ProductNumber,
"price": float(row.ListPrice),
"color": row.Color,
"size": row.Size
} for row in cursor.fetchall()]
@staticmethod
def count(cursor) -> int:
cursor.execute("SELECT COUNT(*) FROM SalesLT.Product")
return cursor.fetchval()
@staticmethod
def create(cursor, product: ProductCreate) -> dict:
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.Size
VALUES (%(name)s, %(product_number)s, %(price)s, %(color)s, %(size)s, %(category_id)s, 0, GETDATE())
""", {
"name": product.name,
"product_number": product.product_number,
"price": product.price,
"color": product.color,
"size": product.size,
"category_id": product.category_id
})
row = cursor.fetchone()
return {
"id": row.ProductID,
"name": row.Name,
"product_number": row.ProductNumber,
"price": float(row.ListPrice),
"color": row.Color,
"size": row.Size
}
@staticmethod
def update(cursor, product_id: int, product: ProductUpdate) -> Optional[dict]:
# Build dynamic update
updates = []
params = {"id": product_id}
if product.name is not None:
updates.append("Name = %(name)s")
params["name"] = product.name
if product.product_number is not None:
updates.append("ProductNumber = %(product_number)s")
params["product_number"] = product.product_number
if product.price is not None:
updates.append("ListPrice = %(price)s")
params["price"] = product.price
if product.category_id is not None:
updates.append("ProductCategoryID = %(category_id)s")
params["category_id"] = product.category_id
if not updates:
return ProductCRUD.get(cursor, product_id)
cursor.execute(f"""
UPDATE SalesLT.Product SET {', '.join(updates)}
OUTPUT INSERTED.ProductID, INSERTED.Name, INSERTED.ProductNumber,
INSERTED.ListPrice, INSERTED.Color, INSERTED.Size
WHERE ProductID = %(id)s
""", params)
row = cursor.fetchone()
if row:
return {
"id": row.ProductID,
"name": row.Name,
"product_number": row.ProductNumber,
"price": float(row.ListPrice),
"color": row.Color,
"size": row.Size
}
return None
@staticmethod
def delete(cursor, product_id: int) -> bool:
cursor.execute("""
DELETE FROM SalesLT.Product WHERE ProductID = %(id)s
""", {"id": product_id})
return cursor.rowcount > 0
@staticmethod
def search(cursor, query: str, skip: int = 0, limit: int = 100) -> List[dict]:
cursor.execute("""
SELECT ProductID, Name, ProductNumber, ListPrice, Color, Size
FROM SalesLT.Product
WHERE Name LIKE %(query)s OR ProductNumber LIKE %(query)s
ORDER BY ProductID
OFFSET %(skip)s ROWS
FETCH NEXT %(limit)s ROWS ONLY
""", {"query": f"%{query}%", "skip": skip, "limit": limit})
return [{
"id": row.ProductID,
"name": row.Name,
"product_number": row.ProductNumber,
"price": float(row.ListPrice),
"color": row.Color,
"size": row.Size
} for row in cursor.fetchall()]
FastAPI 应用
创造 main.py
主模块把所有东西接在一起。 每条路由声明 cursor = Depends(get_db_dependency)，告诉 FastAPI 调用生成器，将获得的光标传递给处理器，随后进行清理。 FastAPI 还会在处理函数运行之前，根据你的 Pydantic 模式对请求体进行验证。
# main.py
from fastapi import FastAPI, HTTPException, Depends, Query
from typing import List
from database import get_db_dependency
from schemas import Product, ProductCreate, ProductUpdate, PaginatedResponse
from crud import ProductCRUD
app = FastAPI(
title="Product API",
description="REST API for products using mssql-python",
version="1.0.0"
)
@app.get("/")
def root():
return {"message": "Product API", "docs": "/docs"}
@app.get("/products", response_model=PaginatedResponse)
def list_products(
page: int = Query(1, ge=1),
page_size: int = Query(10, ge=1, le=100),
cursor = Depends(get_db_dependency)
):
"""List all products with pagination."""
skip = (page - 1) * page_size
items = ProductCRUD.get_all(cursor, skip=skip, limit=page_size)
total = ProductCRUD.count(cursor)
return {
"items": items,
"total": total,
"page": page,
"page_size": page_size,
"pages": (total + page_size - 1) // page_size
}
@app.get("/products/{product_id}", response_model=Product)
def get_product(product_id: int, cursor = Depends(get_db_dependency)):
"""Get a specific product by ID."""
product = ProductCRUD.get(cursor, product_id)
if not product:
raise HTTPException(status_code=404, detail="Product not found")
return product
@app.post("/products", response_model=Product, status_code=201)
def create_product(product: ProductCreate, cursor = Depends(get_db_dependency)):
"""Create a new product."""
return ProductCRUD.create(cursor, product)
@app.put("/products/{product_id}", response_model=Product)
def update_product(
product_id: int,
product: ProductUpdate,
cursor = Depends(get_db_dependency)
):
"""Update an existing product."""
updated = ProductCRUD.update(cursor, product_id, product)
if not updated:
raise HTTPException(status_code=404, detail="Product not found")
return updated
@app.delete("/products/{product_id}", status_code=204)
def delete_product(product_id: int, cursor = Depends(get_db_dependency)):
"""Delete a product."""
if not ProductCRUD.delete(cursor, product_id):
raise HTTPException(status_code=404, detail="Product not found")
@app.get("/products/search/", response_model=List[Product])
def search_products(
q: str = Query(..., min_length=1),
page: int = Query(1, ge=1),
page_size: int = Query(10, ge=1, le=100),
cursor = Depends(get_db_dependency)
):
"""Search products by name or product number."""
skip = (page - 1) * page_size
return ProductCRUD.search(cursor, q, skip=skip, limit=page_size)
# Health check endpoint
@app.get("/health")
def health_check(cursor = Depends(get_db_dependency)):
"""Check database connectivity."""
try:
cursor.execute("SELECT 1")
return {"status": "healthy", "database": "connected"}
except Exception:
raise HTTPException(status_code=503, detail="Database unavailable")
运行应用程序
uvicorn main:app --reload --host 0.0.0.0 --port 8000
服务器在 http://localhost:8000 上侦听。 在你运行API时，保持这个终端运行。
调用 API
在浏览器中打开 http://localhost:8000/docs。 FastAPI 为每条路由显示交互式文档。
- 展开 GET /health，选择 “试用”，然后选择 执行。 确认响应有状态代码
200并报告数据库连接正常。 - 展开GET /products，选择“试用”，将
page_size设置为5，然后选择“执行”。 响应包含五种产品及分页详情。 - 复制
id响应中的一个数值。 展开 GET /products/{product_id}，选择 “试用”，输入复制的值product_id，然后选择 执行。 - 展开 GET /products/search/，选择 “试用”，输入搜索词（如
bikeforq），然后选择 执行。
测试并部署应用
有关错误处理、连接池、认证、测试和部署的指导，请参见 “用mssql-python测试和部署FastAPI应用”。