在 mssql-python 中使用 JSON 資料

Microsoft SQL 2016 及以後版本與 Azure SQL 透過操作nvarchar欄位的函式提供 JSON 支援。 mssql-python 驅動程式以一般 Python 字串形式傳送與接收 JSON。 您可以：

• 將 JSON 以字串形式儲存在 nvarchar 欄位中。
• 使用 JSON_VALUE、JSON_QUERY 和 OPENJSON 的路徑運算式查詢 JSON。
• 將關聯資料轉換為 JSON 格式。FOR JSON
• 將 JSON 解析成關聯格式，使用 OPENJSON。

Note

Microsoft SQL 將 JSON 資料儲存在nvarchar欄位中，而非專用的 JSON 欄位類型。 mssql-python 驅動程式以一般字串形式傳送與接收 JSON。 使用 Python 內建的模組，在用戶端進行json序列化和反序列化。

儲存 JSON 資料

先將 Python 字典序列化為字串json.dumps()，再插入至 nvarchar 欄位。

插入 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 屬性

從 JSON 物件中刪除屬性時，將其設為 NULL：

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

將關聯式查詢結果轉換成 Python 格式，方法是將每一列轉成字典，再序列化成 JSON。

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

透過為無資料的欄位插入 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}")

相關內容

• 數據類型對應
• 字串與 Unicode 處理
• Microsoft SQL JSON 文件