在 mssql-python 中使用 XML 数据

Microsoft SQL 提供了一种具有服务器端处理能力的本地xml数据类型：

XQuery用于查询XML内容。
用于提取和修改的XML方法（value()， query()， exist()nodes()modify()）。
可选的XML模式验证。
用于提高性能的 XML 索引

mssql-python驱动以字符串形式发送和接收XML数据。所有XML处理（XQuery、XML方法、模式验证）都在Microsoft SQL端执行。使用Pythonxml.etree.ElementTree或类似库进行客户端XML解析。

何时使用XML与JSON： 当你需要模式验证、命名空间支持或混合内容（文本与元素交错）时，可以使用 XML。当你的数据以键值对为导向、供 Web API 使用，或不需要强制实施架构约束时，请使用 JSON（配合 nvarchar 列和 Microsoft SQL 的 JSON 函数）。大多数新应用更倾向于 JSON 格式，除非数据本身是文档结构的。

插入 XML 数据

将 XML 传递为 Python 字符串;驱动程序将其发送到 Microsoft SQL 的原生 XML 列类型。

作为字符串插入

通过参数化查询，直接将XML内容作为字符串插入XML列。

import mssql_python
from xml.etree import ElementTree as ET

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

# Create temp table for XML storage
cursor.execute("CREATE TABLE #XMLOrders (OrderID INT IDENTITY(1,1) PRIMARY KEY, OrderXML XML)")

# Insert XML as string
xml_data = """
<Order OrderID="1001">
    <Customer Name="John Doe" Email="john@example.com"/>
    <Items>
        <Item ProductID="A1" Quantity="2" Price="29.99"/>
        <Item ProductID="B2" Quantity="1" Price="49.99"/>
    </Items>
</Order>
"""

cursor.execute("""
    INSERT INTO #XMLOrders (OrderXML) VALUES (%(xml)s)
""", {"xml": xml_data})
conn.commit()

用 ElementTree 构建 XML

使用Python的ElementTree库程序生成XML文档，然后转换为字符串进行插入。

from xml.etree import ElementTree as ET

# Build XML document
order = ET.Element("Order", OrderID="1002")
customer = ET.SubElement(order, "Customer", Name="Jane Smith", Email="jane@example.com")
items = ET.SubElement(order, "Items")
ET.SubElement(items, "Item", ProductID="C3", Quantity="3", Price="19.99")
ET.SubElement(items, "Item", ProductID="D4", Quantity="2", Price="39.99")

# Convert to string
xml_string = ET.tostring(order, encoding="unicode")

cursor.execute("""
    INSERT INTO #XMLOrders (OrderXML) VALUES (%(xml)s)
""", {"xml": xml_string})
conn.commit()

插入前验证 XML

在服务器上使用 SQL 的 TRY/CATCH 和 XML 类型铸造验证 XML 语法，以在存储前拒绝格式错误的文档。

# Validate XML with TRY/CATCH
cursor.execute("""
    BEGIN TRY
        DECLARE @xml XML = CAST(%(xml)s AS XML);
        SELECT @xml.value('(/Order/@OrderID)[1]', 'INT') AS Validated;
    END TRY
    BEGIN CATCH
        THROW;
    END CATCH
""", {"xml": xml_data})

查询XML数据

在服务器端使用 XML 方法提取特定值，而无需将完整文档传给客户端。

XML value（）方法

使用 value() XPath表达式的方法，从XML中提取单个标量值或属性。

从 XML 中提取标量值：

cursor.execute("""
    SELECT 
        OrderID,
        OrderXML.value('(/Order/@OrderID)[1]', 'INT') AS XmlOrderID,
        OrderXML.value('(/Order/Customer/@Name)[1]', 'NVARCHAR(100)') AS CustomerName,
        OrderXML.value('(/Order/Customer/@Email)[1]', 'NVARCHAR(100)') AS Email
    FROM #XMLOrders
""")

for row in cursor:
    print(f"Order {row.XmlOrderID}: {row.CustomerName} ({row.Email})")

XML query（）方法

使用XPath表达式返回文档中的XML片段（非标量值）。

提取XML片段：

cursor.execute("""
    SELECT 
        OrderID,
        OrderXML.query('/Order/Items') AS ItemsXML
    FROM #XMLOrders
    WHERE OrderID = %(id)s
""", {"id": 1})

row = cursor.fetchone()
# row.ItemsXML is an XML string
items_xml = row.ItemsXML
print(items_xml)

XML exist（）方法

测试XPath表达式是否匹配文档中的任何节点，返回1表示true，返回0表示false。

检查XPath是否匹配：

cursor.execute("""
    SELECT OrderID, OrderXML
    FROM #XMLOrders
    WHERE OrderXML.exist('/Order/Items/Item[@ProductID="A1"]') = 1
""")

for row in cursor:
    print(f"Order {row.OrderID} contains product A1")

XML nodes() 方法（分解为行）

使用操作 CROSS APPLY 符和 nodes() 方法将嵌套的XML元素转换为关系行集。

将XML转换为关系格式：

cursor.execute("""
    SELECT 
        o.OrderID,
        Items.Item.value('@ProductID', 'VARCHAR(10)') AS ProductID,
        Items.Item.value('@Quantity', 'INT') AS Quantity,
        Items.Item.value('@Price', 'DECIMAL(10,2)') AS Price
    FROM #XMLOrders o
    CROSS APPLY o.OrderXML.nodes('/Order/Items/Item') AS Items(Item)
    WHERE o.OrderID = %(id)s
""", {"id": 1})

for row in cursor:
    print(f"Product {row.ProductID}: {row.Quantity} x ${row.Price}")

修改XML数据

使用 XML.modify() XQuery DML表达式来更新现有的XML内容。

XML modify（）并插入

使用 XQuery 的 insert 操作，通过 modify() 方法向 XML 文档中添加新元素。

cursor.execute("""
    UPDATE #XMLOrders
    SET OrderXML.modify('
        insert <Item ProductID="E5" Quantity="1" Price="59.99"/>
        into (/Order/Items)[1]
    ')
    WHERE OrderID = %(id)s
""", {"id": 1})
conn.commit()

XML modify() 与 delete 一起使用

使用 modify() 方法和 XQuery 的 delete 操作从 XML 文档中删除元素或节点。

cursor.execute("""
    UPDATE #XMLOrders
    SET OrderXML.modify('
        delete /Order/Items/Item[@ProductID="A1"]
    ')
    WHERE OrderID = %(id)s
""", {"id": 1})
conn.commit()

XML modify() 与 replace

使用 modify() 方法和 XQuery 的 replace value of 操作，更新 XML 文档中的属性值或元素文本。

cursor.execute("""
    UPDATE #XMLOrders
    SET OrderXML.modify('
        replace value of (/Order/Items/Item[@ProductID="B2"]/@Price)[1]
        with 54.99
    ')
    WHERE OrderID = %(id)s
""", {"id": 1})
conn.commit()

对于XML查询

在 FOR XML 任意查询中附加以返回单个XML字符串的结果。

适用于XML RAW。

生成 XML，每行都变成一个简单元素，属性为列。

cursor.execute("""
    SELECT ProductID, Name, ListPrice
    FROM Production.Product
    WHERE ProductSubcategoryID = %(cat)s
    FOR XML RAW('Product'), ROOT('Products')
""", {"cat": 1})

xml_result = cursor.fetchval()
print(xml_result)
# <Products><Product ProductID="1" Name="..." ListPrice="..."/></Products>

FOR XML AUTO

生成带有嵌套结构的XML，自动反映查询中的连接层级。

cursor.execute("""
    SELECT sc.Name AS SubcategoryName, p.Name AS ProductName, p.ListPrice
    FROM Production.ProductSubcategory sc
    JOIN Production.Product p ON sc.ProductSubcategoryID = p.ProductSubcategoryID
    WHERE sc.ProductSubcategoryID = %(cat)s
    FOR XML AUTO, ROOT('Catalog')
""", {"cat": 1})

xml_result = cursor.fetchval()
# Nested XML structure based on join hierarchy

对于XML路径

利用显式列别名和子查询构建自定义XML结构，以控制嵌套和元素名称。

对 XML 结构拥有最大控制权：

cursor.execute("""
    SELECT 
        sc.ProductSubcategoryID AS '@ID',
        sc.Name AS 'Name',
        (
            SELECT p.ProductID AS '@ID',
                   p.Name AS 'Name',
                   p.ListPrice AS 'Price'
            FROM Production.Product p
            WHERE p.ProductSubcategoryID = sc.ProductSubcategoryID
            FOR XML PATH('Product'), TYPE
        ) AS 'Products'
    FROM Production.ProductSubcategory sc
    WHERE sc.ProductSubcategoryID = %(cat)s
    FOR XML PATH('Subcategory'), ROOT('Catalog')
""", {"cat": 1})

xml_result = cursor.fetchval()

用 Python 解析 XML

检索 XML 字符串，并用 xml.etree.ElementTree 兼容库解析。

用 ElementTree 解析查询结果

从数据库获取 XML，并用 ElementTree 解析成 Python 对象树，通过 DOM 访问元素和属性。

from xml.etree import ElementTree as ET

cursor.execute("""
    SELECT CatalogDescription FROM Production.ProductModel
    WHERE CatalogDescription IS NOT NULL AND ProductModelID = 19
""")

row = cursor.fetchone()
root = ET.fromstring(row.CatalogDescription)

# Navigate XML structure using namespace
ns = {'pd': 'http://schemas.microsoft.com/sqlserver/2004/07/adventure-works/ProductModelDescription'}
summary = root.find('.//pd:Summary', ns)
if summary is not None:
    # Get all text content
    text = ''.join(summary.itertext()).strip()
    print(f"Summary: {text[:100]}")

将XML转换为词典

将XML树结构转换为嵌套的Python字典，便于程序化访问嵌套数据。

def xml_to_dict(element):
    """Convert XML element to dictionary."""
    result = {}
    
    # Add attributes
    if element.attrib:
        result['@attributes'] = element.attrib
    
    # Add children
    children = list(element)
    if children:
        child_dict = {}
        for child in children:
            child_result = xml_to_dict(child)
            if child.tag in child_dict:
                # Convert to list if multiple same-named children
                if not isinstance(child_dict[child.tag], list):
                    child_dict[child.tag] = [child_dict[child.tag]]
                child_dict[child.tag].append(child_result)
            else:
                child_dict[child.tag] = child_result
        result.update(child_dict)
    
    # Add text content
    if element.text and element.text.strip():
        result['#text'] = element.text.strip()
    
    return result

# Usage
cursor.execute("""
    SELECT CatalogDescription FROM Production.ProductModel
    WHERE CatalogDescription IS NOT NULL AND ProductModelID = 19
""")
xml_string = cursor.fetchval()
root = ET.fromstring(xml_string)
data = xml_to_dict(root)

处理大型XML文档

Microsoft SQL 可能会将大量FOR XML结果拆分到多行;在解析前将部分串接起来。

分块获取 XML

当FOR XML返回大量结果集时，Microsoft SQL 可能会将 XML 拆分到多行;将所有部分合并为单一文档。

# FOR XML might split large results across rows
cursor.execute("""
    SELECT * FROM LargeTable FOR XML RAW
""")

xml_parts = []
for row in cursor:
    xml_parts.append(row[0])

full_xml = "".join(xml_parts)

流式 XML 解析

对于非常大的 XML 文档，可以使用迭代解析，一次处理一个元素，而无需将整棵树加载到内存中。

from xml.etree import ElementTree as ET
import io

cursor.execute("""
    SELECT Instructions FROM Production.ProductModel
    WHERE Instructions IS NOT NULL AND ProductModelID = 7
""")
xml_string = cursor.fetchval()

# Use iterparse for memory-efficient parsing
xml_stream = io.StringIO(xml_string)
for event, elem in ET.iterparse(xml_stream, events=['end']):
    if elem.tag.endswith('step'):
        # Process step element
        text = ''.join(elem.itertext()).strip()
        if text:
            print(f"Step: {text[:60]}")
        # Free memory
        elem.clear()

XML 索引

在 Microsoft SQL 中创建索引以加快 XML 查询速度：

-- Primary XML index (assumes a table with an XML column)
CREATE PRIMARY XML INDEX PIX_XMLOrders_OrderXML
ON #XMLOrders(OrderXML);

-- Secondary indexes for specific access patterns
CREATE XML INDEX SIX_XMLOrders_Path
ON #XMLOrders(OrderXML)
USING XML INDEX PIX_XMLOrders_OrderXML
FOR PATH;

CREATE XML INDEX SIX_XMLOrders_Value
ON #XMLOrders(OrderXML)
USING XML INDEX PIX_XMLOrders_OrderXML
FOR VALUE;

使用命名空间

在 XQuery 表达式中使用 declare namespace 语法以内联方式声明命名空间前缀。

查询带有命名空间的 XML

在XPath表达式中添加命名空间声明，以匹配特定命名空间中的元素。

xml_with_ns = """
<Order xmlns="http://example.com/orders" 
       xmlns:c="http://example.com/customer">
    <c:Customer Name="John Doe"/>
    <Items>
        <Item ProductID="A1"/>
    </Items>
</Order>
"""

cursor.execute("""
    SELECT OrderXML.value('
        declare namespace o="http://example.com/orders";
        declare namespace c="http://example.com/customer";
        (/o:Order/c:Customer/@Name)[1]
    ', 'NVARCHAR(100)') AS CustomerName
    FROM #XMLOrders
    WHERE OrderID = %(id)s
""", {"id": 1})

用 Python 解析命名空间的 XML

在解析 Python 中带有命名空间的 XML 时，请在你find()和其他元素访问调用中声明命名空间映射。

from xml.etree import ElementTree as ET

cursor.execute("""
    SELECT TOP 1 
        (SELECT SalesOrderID AS [@OrderID],
                TotalDue AS [@Total],
                CustomerID AS [Customer/@ID]
         FROM Sales.SalesOrderHeader
         WHERE SalesOrderID = 43659
         FOR XML PATH('Order'))
""")
xml_string = cursor.fetchval()
root = ET.fromstring(xml_string)
order_id = root.get('OrderID')
total = root.get('Total')
print(f"Order {order_id}: ${total}")

最佳做法

应用这些指南，高效地处理XML数据。

选择 XML 而非 JSON

根据你的数据结构和处理需求，决定是使用Microsoft SQL的原生xml类型还是存储JSONnvarchar：

此比较帮助您决定是否使用 Microsoft SQL xml 类型，或将 JSON 存储在nvarchar：

Feature	XML	JSON
架构验证	原生支持	无原生支持
命名空间	完全支持	不支持
特性	支持	无直接等效项
混合内容	支持	不支持
文档处理	更好	结构化程度较低

避免使用 SELECT *

当你只需要特定值时，不要检索完整的 XML 文档。使用XML方法提取服务器上的数据。这种方法减少了网络流量，避免了用 Python 解析大型 XML 文档。

# Create sample XML table for performance demos
cursor.execute("""
    CREATE TABLE #XMLPerf (
        OrderID INT,
        OrderXML XML
    )
""")
cursor.execute("""
    INSERT INTO #XMLPerf (OrderID, OrderXML) VALUES
    (1, '<Order><Customer Name="Alice"/><Item ProductID="1" Qty="2" Price="29.99"/></Order>'),
    (2, '<Order><Customer Name="Bob"/><Item ProductID="3" Qty="1" Price="49.99"/></Order>')
""")

# Anti-pattern: Downloads entire XML per row
cursor.execute("SELECT * FROM #XMLPerf")

# Better: Extract only the values you need on the server
cursor.execute("""
    SELECT OrderID,
           OrderXML.value('(/Order/Customer/@Name)[1]', 'NVARCHAR(100)') AS Customer
    FROM #XMLPerf
""")

处理空 XML

查询 XML 数据时，在 XML 方法返回 NULL 时，使用 COALESCE() 提供默认值。

cursor.execute("""
    SELECT OrderID,
           COALESCE(OrderXML.value('(/Order/@Total)[1]', 'DECIMAL(10,2)'), 0) AS Total
    FROM #XMLPerf
""")

反馈

此页面是否有帮助？

Last updated on 2026-07-22