Utilize dados JSON com mssql-python

O Microsoft SQL 2016 e versões posteriores e o SQL do Azure fornecem suporte JSON através de funções que operam em nvarchar colunas. O driver mssql-python envia e recebe JSON como strings Python normais. É possível:

Armazene JSON como cadeias de caracteres nas colunas nvarchar.
Consulta JSON com expressões de caminho usando JSON_VALUE, JSON_QUERY, e OPENJSON.
Transforme dados relacionais para JSON com FOR JSON.
Analise JSON em formato relacional com OPENJSON.

Note

O Microsoft SQL armazena dados JSON em nvarchar colunas, não num tipo de coluna JSON dedicado. O driver mssql-python envia e recebe JSON como strings regulares. Usa o módulo incorporado json do Python para serializar e desserializar do lado do cliente.

Armazenar dados JSON

Serializar dicionários Python em strings com json.dumps() antes de os inserir em colunas nvarchar.

Inserir cadeia JSON

Armazene um dicionário Python como texto JSON numa tabela de base de dados:

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()

Validar JSON na inserção

Use a ISJSON() função para validar a sintaxe JSON antes de inserir:

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")

Consultar dados JSON

Use funções de caminho JSON SQL Microsoft para extrair valores no servidor antes de os devolver ao cliente.

Extrair valores escalares

Uso JSON_VALUE para extrair valores únicos:

# 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}")

Extrair objetos ou arrays

Use JSON_QUERY para objetos e arrays.

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}")

Converter array JSON em linhas

Expanda um array JSON em linhas usando 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}")

Analisar objeto JSON em colunas

Extrair campos individuais de objetos JSON usando JSON_VALUE() e lançar os resultados para tipos SQL apropriados.

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
""")

Modificar dados JSON

Use JSON_MODIFY para atualizar um caminho específico num documento JSON sem reescrever todo o valor.

Atualizar valor JSON

Modificar uma única propriedade JSON usando JSON_MODIFY:

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()

Adicionar propriedade JSON

Insira uma nova propriedade num objeto JSON existente:

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})

Remover propriedade JSON

Elimine uma propriedade de um objeto JSON definindo-a para 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})

Adicionar ao array JSON

Adicione um novo valor ao final de um array JSON usando append diretiva em JSON_MODIFY:

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})

Converter dados relacionais em JSON

A FOR JSON cláusula transforma os resultados da consulta numa string JSON no lado do servidor.

PARA A JSON AUTO

Gerar JSON a partir dos resultados da consulta:

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))

PARA O CAMINHO JSON

Obtenha mais controlo sobre a estrutura 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 aninhado

Consulte estruturas de dados com matrizes e objetos JSON aninhados, utilizando subconsultas com FOR JSON para construir um resultado JSON hierárquico.

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

Padrões de integração em Python

Estes padrões mostram como construir abstrações em Python sobre tabelas suportadas por JSON.

Padrão de repositório com JSON

Implemente uma camada de acesso a dados que serialize e desserialize objetos Python para colunas JSON, fornecendo uma interface segura para tipos para a base de dados.

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)
        )

Liga-te à base de dados, depois cria o repositório e usa-o para guardar e recuperar um produto. O save() método assume o INSERT ramo quando id é None e o ramo UPDATE caso contrário:

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()

O repositório cria #JsonRepo como uma tabela temporária local no âmbito da conexão que é passada, pelo que save() e get() devem partilhar essa mesma conexão. A tabela é eliminada quando a conexão é encerrada.

Lidar eficientemente com grandes resultados JSON

Quando os resultados em JSON forem grandes, obtenha-os em partes por várias linhas.

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", {})

Converter resultados de consultas para JSON em Python

Transforme os resultados das consultas relacionais para formato JSON em Python convertendo cada linha para um dicionário e depois serializando para 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)

Indexar dados JSON

Crie uma coluna calculada suportada por uma expressão de caminho JSON para tornar o caminho indexável.

Coluna computada com índice

Defina uma coluna calculada que extraia um valor JSON e aplique-lhe um índice para filtragem eficiente nos caminhos JSON frequentemente consultados. O exemplo seguinte cria uma tabela permanente, adiciona uma coluna computada persistente sobre o caminho JSON $.specs.weight e cria um índice sobre ela.

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()

Consulta usando a coluna computada indexada

Filtra diretamente pela coluna calculada. O motor de consulta utiliza o índice em vez de percorrer e analisar todos os documentos 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()

Escolha entre colunas relacionais e armazenamento JSON

Use colunas relacionais quando os dados têm um esquema fixo, necessitam de integridade referencial, participam em JOINs ou aparecem frequentemente em cláusulas WHERE. Use colunas JSON (nvarchar(max)) quando os dados são escassos, variam entre linhas ou representam configurações ou metadados flexíveis.

Quando usar processamento JSON do lado do servidor versus do lado do cliente

Use funções JSON de Microsoft SQL (JSON_VALUE, JSON_QUERY, OPENJSON) quando precisar de filtrar, indexar ou agregar entre campos JSON sem obter todas as linhas para o cliente. Esta escolha é correta quando apenas um subconjunto de linhas corresponde aos seus critérios, ou quando quer índices de colunas calculados em caminhos JSON.

Usa processamento Python do lado do cliente (json.loads()) quando recuperares documentos inteiros e os processares em lógica de aplicação. Esta abordagem funciona bem quando precisas do documento completo e não filtras campos JSON na base de dados.

Fluxos de trabalho ao estilo de documento

Quando a sua aplicação armazena e recupera documentos inteiros, use serialização do lado Python e trate a coluna JSON como armazenamento opaco. Processe e consulte os documentos em Python buscando e desserializando blobs JSON completos:

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

Consultas JSON do lado do servidor

Usa funções JSON SQL do Microsoft quando precisares de filtrar, indexar ou agregar entre campos JSON sem teres de buscar todas as linhas. Esta abordagem é mais eficiente do que carregar todas as linhas para o Python para filtrar em memória:

JSON_VALUE extrai valores escalares e pode recalcular índices de colunas.
JSON_QUERY extrai objetos e matrizes.
OPENJSON expande o JSON em linhas para operações de JOIN e agregação.
JSON_MODIFY atualiza caminhos específicos sem reescrever todo o documento.

# 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"})

Para caminhos JSON frequentemente consultados, crie uma coluna calculada com um índice:

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

CREATE INDEX IX_Settings_Department ON Settings(Department);

Melhores práticas

Aplique estas orientações para usar colunas JSON de forma fiável.

Validar JSON antes do armazenamento

Valide JSON e identificadores de tabela/coluna antes de o armazenar para evitar ataques por injeção.

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})

Não abuses do JSON

Use colunas JSON para dados flexíveis ou esparsos, como preferências do utilizador ou campos personalizados. Use colunas relacionais para:

Dados frequentemente consultados.
Dados que precisam de integridade referencial.
Colunas usadas nas WHERE orações.

Lidar corretamente com Nenhum/NULO

Trate os campos JSON em falta ou opcionais, inserindo valores NULL nas colunas sem dados.

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}")

Comentários

Esta página foi útil?

Last updated on 2026-07-22