在 mssql-python 中使用 JSON 数据

Microsoft SQL 2016及以后版本和Azure SQL通过操作nvarchar列的函数支持JSON。 mssql-python 驱动以常规 Python 字符串的形式发送和接收 JSON。 您可以：

  • 在 nvarchar 列中将 JSON 存储为字符串。
  • 使用 JSON_VALUE、JSON_QUERY 和 OPENJSON 通过路径表达式查询 JSON。
  • 使用 FOR JSON 将关系数据转换为 JSON。
  • 将 JSON 解析为关系格式，使用 OPENJSON。

注释

Microsoft SQL 将 JSON 数据存储在nvarchar列中，而非专用的 JSON 列类型。 mssql-python驱动以常规字符串形式发送和接收JSON。 使用 Python 的内置 json 模块在客户端进行序列化和反序列化。

存储JSON数据

在将 Python 字典插入到 nvarchar 列之前，先使用 json.dumps() 将其序列化为字符串。

插入 JSON 字符串

将 Python 词典以 JSON 文本存储在数据库表中：

import json
import mssql_python

conn = mssql_python.connect(
    "Server=<server>.database.windows.net;"
    "Database=<database>;"
    "Authentication=ActiveDirectoryDefault;"
    "Encrypt=yes"
)
cursor = conn.cursor()

# Create table with JSON column
cursor.execute("""
    CREATE TABLE #JsonProducts (
        ProductID INT IDENTITY PRIMARY KEY,
        Name NVARCHAR(100),
        JsonData NVARCHAR(MAX)
    )
""")

# Python dict to JSON string
product_data = {
    "name": "Widget Pro",
    "specs": {
        "weight": 2.5,
        "dimensions": {"width": 10, "height": 5, "depth": 3}
    },
    "tags": ["electronics", "gadgets", "bestseller"]
}

cursor.execute("""
    INSERT INTO #JsonProducts (Name, JsonData)
    VALUES (%(name)s, %(json)s)
""", {"name": "Widget Pro", "json": json.dumps(product_data)})
conn.commit()

插入时验证 JSON

在插入前，使用该 ISJSON() 函数验证 JSON 语法：

data = {"name": "Widget", "specs": {"weight": 1.5}}

cursor.execute("""
    CREATE TABLE #JsonValidate (
        ProductID INT IDENTITY, Name NVARCHAR(100), JsonData NVARCHAR(MAX)
    )
""")
cursor.execute("""
    INSERT INTO #JsonValidate (Name, JsonData)
    SELECT %(name)s, %(json)s
    WHERE ISJSON(%(json)s) = 1
""", {"name": "Widget", "json": json.dumps(data)})

if cursor.rowcount == 0:
    raise ValueError("Invalid JSON data")

查询 JSON 数据

使用 Microsoft SQL JSON 路径函数在返回客户端前提取服务器上的数值。

提取标量值

使用 JSON_VALUE 提取单个值：

# Create table with sample JSON data
cursor.execute("""
    CREATE TABLE #JsonExtract (
        ProductID INT IDENTITY, Name NVARCHAR(100), JsonData NVARCHAR(MAX)
    )
""")
cursor.execute("""
    INSERT INTO #JsonExtract (Name, JsonData) VALUES (
        'Widget Pro',
        '{"name":"Widget Pro","specs":{"weight":2.5,"dimensions":{"width":10,"height":5,"depth":3}},"tags":["electronics","gadgets"]}'
    )
""")

cursor.execute("""
    SELECT 
        Name,
        JSON_VALUE(JsonData, '$.specs.weight') AS Weight,
        JSON_VALUE(JsonData, '$.specs.dimensions.width') AS Width
    FROM #JsonExtract
    WHERE JSON_VALUE(JsonData, '$.name') = %(name)s
""", {"name": "Widget Pro"})

row = cursor.fetchone()
print(f"Weight: {row.Weight}, Width: {row.Width}")

提取对象或数组

对对象和数组使用 JSON_QUERY。

cursor.execute("""
    CREATE TABLE #JsonQuery (
        ProductID INT IDENTITY, Name NVARCHAR(100), JsonData NVARCHAR(MAX)
    )
""")
cursor.execute("""
    INSERT INTO #JsonQuery (Name, JsonData) VALUES (
        'Widget Pro',
        '{"specs":{"weight":2.5,"color":"blue"},"tags":["electronics","gadgets"]}'
    )
""")

cursor.execute("""
    SELECT 
        Name,
        JSON_QUERY(JsonData, '$.specs') AS Specs,
        JSON_QUERY(JsonData, '$.tags') AS Tags
    FROM #JsonQuery
""")

for row in cursor:
    specs = json.loads(row.Specs) if row.Specs else {}
    tags = json.loads(row.Tags) if row.Tags else []
    print(f"{row.Name}: {specs}, Tags: {tags}")

将 JSON 数组解析为行

将JSON数组展开为行，使用：OPENJSON

cursor.execute("""
    CREATE TABLE #JsonArray (
        ProductID INT IDENTITY, Name NVARCHAR(100), JsonData NVARCHAR(MAX)
    )
""")
cursor.execute("""
    INSERT INTO #JsonArray (Name, JsonData) VALUES 
        ('Widget Pro', '{"tags":["electronics","gadgets","bestseller"]}'),
        ('Gadget X', '{"tags":["tools","gadgets"]}')
""")

cursor.execute("""
    SELECT p.Name, t.value AS Tag
    FROM #JsonArray p
    CROSS APPLY OPENJSON(p.JsonData, '$.tags') t
""")

for row in cursor:
    print(f"Product: {row.Name}, Tag: {row.Tag}")

将 JSON 对象解析为列

从JSON JSON_VALUE() 对象中提取单个字段，并转换为相应的SQL类型。

cursor.execute("""
    CREATE TABLE #JsonCols (
        ProductID INT IDENTITY, Name NVARCHAR(100), JsonData NVARCHAR(MAX)
    )
""")
cursor.execute("""
    INSERT INTO #JsonCols (Name, JsonData) VALUES (
        'Widget Pro',
        '{"name":"Widget Pro","specs":{"weight":2.5,"dimensions":{"width":10,"height":5}}}'
    )
""")

cursor.execute("""
    SELECT 
        p.ProductID,
        j.name AS ProductName,
        j.weight,
        j.width,
        j.height
    FROM #JsonCols p
    CROSS APPLY OPENJSON(p.JsonData)
    WITH (
        name NVARCHAR(100) '$.name',
        weight DECIMAL(5,2) '$.specs.weight',
        width INT '$.specs.dimensions.width',
        height INT '$.specs.dimensions.height'
    ) j
""")

修改JSON数据

用于 JSON_MODIFY 更新 JSON 文档中的特定路径，而无需重写整个值。

更新 JSON 值

使用以下 JSON_MODIFY方法修改单个 JSON 属性：

cursor.execute("""
    CREATE TABLE #JsonMod (
        ProductID INT IDENTITY, Name NVARCHAR(100), JsonData NVARCHAR(MAX)
    )
""")
cursor.execute("""
    INSERT INTO #JsonMod (Name, JsonData) VALUES (
        'Widget Pro',
        '{"specs":{"weight":2.5,"dimensions":{"width":10}},"tags":["electronics"]}'
    )
""")

cursor.execute("""
    UPDATE #JsonMod
    SET JsonData = JSON_MODIFY(JsonData, '$.specs.weight', %(weight)s)
    WHERE ProductID = %(id)s
""", {"weight": 3.0, "id": 1})
conn.commit()

添加 JSON 属性

在现有的 JSON 对象中插入一个新属性：

cursor.execute("""
    CREATE TABLE #JsonAdd (
        ProductID INT IDENTITY, Name NVARCHAR(100), JsonData NVARCHAR(MAX)
    )
""")
cursor.execute("""
    INSERT INTO #JsonAdd (Name, JsonData) VALUES (
        'Widget Pro', '{"specs":{"weight":2.5}}'
    )
""")

cursor.execute("""
    UPDATE #JsonAdd
    SET JsonData = JSON_MODIFY(JsonData, '$.specs.color', %(color)s)
    WHERE ProductID = %(id)s
""", {"color": "blue", "id": 1})

移除 JSON 属性

通过将属性设置为 NULL，从 JSON 对象中删除该属性：

cursor.execute("""
    CREATE TABLE #JsonRem (
        ProductID INT IDENTITY, Name NVARCHAR(100), JsonData NVARCHAR(MAX)
    )
""")
cursor.execute("""
    INSERT INTO #JsonRem (Name, JsonData) VALUES (
        'Widget Pro', '{"specs":{"weight":2.5,"color":"blue"}}'
    )
""")

cursor.execute("""
    UPDATE #JsonRem
    SET JsonData = JSON_MODIFY(JsonData, '$.specs.color', NULL)
    WHERE ProductID = %(id)s
""", {"id": 1})

追加到 JSON 数组

在 JSON_MODIFY 中使用 append 指令向 JSON 数组的末尾添加一个新值：

cursor.execute("""
    CREATE TABLE #JsonAppend (
        ProductID INT IDENTITY, Name NVARCHAR(100), JsonData NVARCHAR(MAX)
    )
""")
cursor.execute("""
    INSERT INTO #JsonAppend (Name, JsonData) VALUES (
        'Widget Pro', '{"tags":["electronics","gadgets"]}'
    )
""")

cursor.execute("""
    UPDATE #JsonAppend
    SET JsonData = JSON_MODIFY(
        JsonData, 
        'append $.tags', 
        %(tag)s
    )
    WHERE ProductID = %(id)s
""", {"tag": "new-arrival", "id": 1})

将关系数据转换为 JSON

该 FOR JSON 条款在服务器端将查询结果转换为 JSON 字符串。

FOR JSON AUTO

从查询结果生成 JSON：

cursor.execute("""
    SELECT TOP 5 o.SalesOrderID, p.LastName AS CustomerName, o.TotalDue
    FROM Sales.SalesOrderHeader o
    JOIN Sales.Customer c ON o.CustomerID = c.CustomerID
    JOIN Person.Person p ON c.PersonID = p.BusinessEntityID
    FOR JSON AUTO
""")

# Result is a single string containing JSON
json_result = cursor.fetchval()
orders = json.loads(json_result)
print(json.dumps(orders, indent=2))

FOR JSON PATH

更好地控制JSON结构：

cursor.execute("""
    SELECT 
        o.SalesOrderID AS 'order.id',
        o.OrderDate AS 'order.date',
        p.LastName AS 'customer.name',
        e.EmailAddress AS 'customer.email'
    FROM Sales.SalesOrderHeader o
    JOIN Sales.Customer c ON o.CustomerID = c.CustomerID
    JOIN Person.Person p ON c.PersonID = p.BusinessEntityID
    JOIN Person.EmailAddress e ON p.BusinessEntityID = e.BusinessEntityID
    WHERE o.SalesOrderID = %(id)s
    FOR JSON PATH, WITHOUT_ARRAY_WRAPPER
""", {"id": 43659})

json_result = cursor.fetchval()
order = json.loads(json_result)
# Structure: {"order": {"id": 43659, "date": "..."}, "customer": {"name": "...", "email": "..."}}

嵌套 JSON

使用带有 FOR JSON 的子查询来构建层次化 JSON 输出，并查询包含嵌套 JSON 数组和对象的数据结构。

cursor.execute("""
    SELECT TOP 3
        c.CustomerID,
        p.LastName AS CustomerName,
        (SELECT TOP 3 o.SalesOrderID, o.TotalDue
         FROM Sales.SalesOrderHeader o
         WHERE o.CustomerID = c.CustomerID
         FOR JSON PATH) AS Orders
    FROM Sales.Customer c
    JOIN Person.Person p ON c.PersonID = p.BusinessEntityID
    WHERE c.PersonID IS NOT NULL
    FOR JSON PATH
""")

json_result = cursor.fetchval()
customers = json.loads(json_result)
# Each customer has nested Orders array

Python 集成模式

这些模式展示了如何在 JSON 支持的表上构建 Python 抽象。

带有 JSON 的仓库模式

实现一个数据访问层，将 Python 对象序列化和反序列化为 JSON 列，提供数据库的类型安全接口。

from dataclasses import dataclass, asdict
from typing import Optional
import json

@dataclass
class ProductSpecs:
    weight: float
    color: str
    dimensions: dict

@dataclass
class Product:
    id: Optional[int]
    name: str
    specs: ProductSpecs

class ProductRepository:
    def __init__(self, connection):
        self.conn = connection
        cursor = self.conn.cursor()
        cursor.execute("""
            IF OBJECT_ID('#JsonRepo') IS NULL
                CREATE TABLE #JsonRepo (
                    ProductID INT IDENTITY PRIMARY KEY,
                    Name NVARCHAR(100),
                    JsonData NVARCHAR(MAX)
                )
        """)
        self.conn.commit()
    
    def save(self, product: Product) -> int:
        cursor = self.conn.cursor()
        specs_json = json.dumps(asdict(product.specs))
        
        if product.id:
            cursor.execute("""
                UPDATE #JsonRepo SET Name = %(name)s, JsonData = %(json)s
                WHERE ProductID = %(id)s
            """, {"name": product.name, "json": specs_json, "id": product.id})
        else:
            cursor.execute("""
                INSERT INTO #JsonRepo (Name, JsonData)
                OUTPUT INSERTED.ProductID
                VALUES (%(name)s, %(json)s)
            """, {"name": product.name, "json": specs_json})
            product.id = cursor.fetchval()
        
        self.conn.commit()
        return product.id
    
    def get(self, product_id: int) -> Optional[Product]:
        cursor = self.conn.cursor()
        cursor.execute("""
            SELECT ProductID, Name, JsonData FROM #JsonRepo WHERE ProductID = %(id)s
        """, {"id": product_id})
        
        row = cursor.fetchone()
        if row is None:
            return None
        
        specs_data = json.loads(row.JsonData)
        return Product(
            id=row.ProductID,
            name=row.Name,
            specs=ProductSpecs(**specs_data)
        )

连接到数据库，然后创建仓库，用它来保存和检索产品。 save()方法在id为None时采用INSERT分支，否则采用UPDATE分支：

conn = mssql_python.connect(connection_string)

repo = ProductRepository(conn)

# id is None, so save() inserts a new row and returns the generated ProductID.
product = Product(
    id=None,
    name="Widget Pro",
    specs=ProductSpecs(weight=2.5, color="black", dimensions={"width": 10, "height": 5})
)
product_id = repo.save(product)
print(f"Saved product {product_id}")

# Read the product back into a typed Product object.
loaded = repo.get(product_id)
print(loaded)

conn.close()

该仓储会将 #JsonRepo 创建为一个本地临时表，其作用域仅限于你传入的连接，因此 save() 和 get() 必须共享同一个连接。 连接关闭时，该表将被删除。

高效处理大型 JSON 结果

当 JSON 结果较大时，可分部分跨多行获取。

def fetch_json_in_parts(cursor, query: str, params: dict) -> list:
    """Handle JSON results that might span multiple rows."""
    cursor.execute(query, params)
    
    # FOR JSON might split large results across rows
    json_parts = []
    for row in cursor:
        json_parts.append(row[0])
    
    # Combine parts
    json_string = "".join(json_parts)
    return json.loads(json_string) if json_string else []

# Usage
data = fetch_json_in_parts(cursor, "SELECT TOP 100 * FROM Production.Product FOR JSON AUTO", {})

将查询结果转换为 Python 中的 JSON

通过将每行转换为字典，然后序列化为 JSON，将关系查询结果转换为 Python 格式。

def query_to_json(cursor, query: str, params: dict = None) -> str:
    """Execute query and return results as JSON string."""
    cursor.execute(query, params or {})
    columns = [col[0] for col in cursor.description]
    
    rows = []
    for row in cursor:
        rows.append(dict(zip(columns, row)))
    
    return json.dumps(rows, default=str, indent=2)

# Usage
json_output = query_to_json(cursor, "SELECT TOP 5 ProductID, Name, ListPrice FROM Production.Product WHERE ProductSubcategoryID = %(cat)s", {"cat": 1})
print(json_output)

索引 JSON 数据

创建一个计算出的列，并用 JSON 路径表达式支持，使路径可索引。

具有索引的计算列

定义一个计算出的列，提取 JSON 值，并为其应用索引，以便高效过滤频繁查询的 JSON 路径。 以下示例创建一个永久表，在 JSON $.specs.weight 路径上添加持久化的计算列，并在其上创建索引。

cursor.execute("""
    IF OBJECT_ID('dbo.ProductCatalog', 'U') IS NOT NULL
        DROP TABLE dbo.ProductCatalog
""")
cursor.execute("""
    CREATE TABLE dbo.ProductCatalog (
        ProductID   INT IDENTITY PRIMARY KEY,
        Name        NVARCHAR(100),
        JsonData    NVARCHAR(MAX)
    )
""")

# Insert sample rows with JSON data
rows = [
    ("Widget Pro",   '{"specs":{"weight":2.5,"color":"blue"}}'),
    ("Gadget X",     '{"specs":{"weight":0.8,"color":"red"}}'),
    ("Heavy Duty",   '{"specs":{"weight":9.1,"color":"gray"}}'),
]
cursor.executemany(
    "INSERT INTO dbo.ProductCatalog (Name, JsonData) VALUES (%(name)s, %(json)s)",
    [{"name": n, "json": j} for n, j in rows]
)
conn.commit()

# Add a persisted computed column that extracts weight from JSON
cursor.execute("""
    ALTER TABLE dbo.ProductCatalog
    ADD ProductWeight AS CAST(JSON_VALUE(JsonData, '$.specs.weight') AS DECIMAL(5,2)) PERSISTED
""")

# Index the computed column for efficient range queries
cursor.execute("""
    CREATE INDEX IX_ProductCatalog_Weight
    ON dbo.ProductCatalog (ProductWeight)
""")
conn.commit()

使用已建立索引的计算列进行查询

直接用计算出的列来过滤。 查询引擎使用索引，而不是扫描和解析每一份JSON文档。

cursor.execute("""
    SELECT Name, ProductWeight
    FROM dbo.ProductCatalog
    WHERE ProductWeight > %(min_weight)s
    ORDER BY ProductWeight
""", {"min_weight": 1.0})

for row in cursor:
    print(f"{row.Name}: {row.ProductWeight} kg")

# Cleanup
cursor.execute("DROP TABLE dbo.ProductCatalog")
conn.commit()

在关系列和 JSON 存储之间选择

当数据具有固定架构、需要参照完整性、参与 JOIN 操作或经常出现在 WHERE 子句中时，请使用关系型列。 当数据稀疏、行间变化或表示灵活配置或元数据时，使用JSON列（nvarchar(max)）。

何时使用服务器端与客户端 JSON 处理

当你需要在JSON字段之间进行筛选、索引或聚合，而不必将每一行都导入客户端时，可以使用Microsoft SQL JSON函数（JSON_VALUE， JSON_QUERY， OPENJSON）。 当只有部分行符合你的条件，或者你想在JSON路径上计算出列索引时，这个选择是正确的。

在检索整份文档并用应用逻辑处理时，使用客户端的 Python 处理（json.loads()）。 这种方法在你需要完整文档且不在数据库中筛选 JSON 字段时效果很好。

文档式工作流程

当你的应用存储和检索整份文档时，使用Python端的序列化，并将JSON列视为不透明存储。 通过获取和反序列化完整的 JSON blob，处理和查询 Python 文档：

import json

# Create the settings table
cursor.execute("""
    CREATE TABLE #Settings (
        UserID INT PRIMARY KEY,
        ConfigJson NVARCHAR(MAX)
    )
""")

# Store a configuration document
config = {
    "theme": "dark",
    "notifications": {"email": True, "sms": False},
    "custom_fields": {"department": "Engineering", "cost_center": "CC-100"}
}

cursor.execute(
    "INSERT INTO #Settings (UserID, ConfigJson) VALUES (%(uid)s, %(cfg)s)",
    {"uid": 1, "cfg": json.dumps(config)}
)

# Retrieve and process in Python
cursor.execute("SELECT ConfigJson FROM #Settings WHERE UserID = %(uid)s", {"uid": 1})
row = cursor.fetchone()
config = json.loads(row.ConfigJson)
print(config["notifications"]["email"])  # True

服务器端 JSON 查询

当你需要在JSON字段之间进行筛选、索引或聚合时，可以使用Microsoft SQL JSON函数，而不是每一行都抓取。 这种方法比将所有行加载到 Python 中以进行内存过滤更高效：

  • JSON_VALUE 提取标量值，并可作为计算列索引的基础。
  • JSON_QUERY 提取对象和数组。
  • OPENJSON 将 JSON 解包为多行，用于 JOIN 和聚合。
  • JSON_MODIFY 更新特定路径，无需重写整个文档。
# Filter by a JSON field server-side
cursor.execute("""
    SELECT UserID, ConfigJson
    FROM #Settings
    WHERE JSON_VALUE(ConfigJson, '$.custom_fields.department') = %(dept)s
""", {"dept": "Engineering"})

对于经常查询的JSON路径，创建一个带有索引的计算列：

ALTER TABLE Settings
ADD Department AS JSON_VALUE(ConfigJson, '$.custom_fields.department');

CREATE INDEX IX_Settings_Department ON Settings(Department);

最佳做法

应用这些指南以可靠地使用JSON列。

在存储前验证 JSON

存储前验证 JSON 和表/列标识符，以防止注入攻击。

def store_json_safely(cursor, table: str, json_column: str, data: dict):
    """Store JSON with validation."""
    # Validate identifiers to prevent SQL injection
    import re
    if not re.match(r'^[A-Za-z_][A-Za-z0-9_.]*$', table):
        raise ValueError(f"Invalid table name: {table}")
    if not re.match(r'^[A-Za-z_][A-Za-z0-9_]*$', json_column):
        raise ValueError(f"Invalid column name: {json_column}")

    json_str = json.dumps(data)
    
    # Check if valid JSON in Microsoft SQL
    cursor.execute("SELECT ISJSON(%(json)s)", {"json": json_str})
    if cursor.fetchval() != 1:
        raise ValueError("Invalid JSON")
    
    cursor.execute(f"INSERT INTO {table} ({json_column}) VALUES (%(json)s)", {"json": json_str})

不要过度使用 JSON

对于灵活或稀疏的数据，如用户偏好或自定义字段，可以使用 JSON 列。 关系列用于：

  • 经常被查询的数据。
  • 需要参照完整性的数据。
  • WHERE 子句中使用的列。

正确处理 None/NULL

通过对没有数据的列插入空值来处理缺少或可选的JSON字段。

cursor.execute("""
    CREATE TABLE #JsonOpt (
        ProductID INT IDENTITY, Name NVARCHAR(100), JsonData NVARCHAR(MAX)
    )
""")
cursor.execute("""
    INSERT INTO #JsonOpt (Name, JsonData) VALUES (
        'Widget Pro', '{"required_field":"value"}'
    )
""")

cursor.execute("""
    SELECT 
        Name,
        JSON_VALUE(JsonData, '$.optional_field') AS OptionalValue
    FROM #JsonOpt
""")

for row in cursor:
    # JSON_VALUE returns NULL if path doesn't exist
    value = row.OptionalValue or "default"
    print(f"{row.Name}: {value}")

相关内容

  • Last updated on