Použijte JSON data s mssql-python

Microsoft SQL 2016 a novější verze a Azure SQL poskytují podporu JSON prostřednictvím funkcí pracujících na nvarchar sloupcích. Ovladač mssql-python odesílá a přijímá JSON jako běžné Python řetězce. Máte tyto možnosti:

• Ukládejte JSON jako řetězce ve nvarchar sloupcích.
• Dotazování JSONu pomocí výrazů cesty s použitím JSON_VALUE, JSON_QUERY a OPENJSON.
• Transformujte relační data do JSON pomocí FOR JSON.
• Parsujte JSON do relačního formátu pomocí OPENJSON.

Note

Microsoft SQL ukládá JSON data ve nvarchar sloupcích, nikoli ve vyhrazeném typu JSON sloupce. Ovladač mssql-python odesílá a přijímá JSON jako běžné řetězce. Použijte vestavěný json modul Python pro serializaci a deserializaci na straně klienta.

Uložit JSON data

Před vložením do sloupců nvarchar serializujte slovníky Pythonu na řetězce pomocí json.dumps().

Vložte JSON řetězec

Uložit Python slovník jako JSON text do databázové tabulky:

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

Validujte JSON při vložení

Použijte ISJSON() funkci k ověření syntaxe JSON před vložením:

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

Dotazování dat JSON

Použijte Microsoft SQL JSON path functions k extrakci hodnot na serveru před jejich vrácením klientovi.

Extrahujte skalární hodnoty

Použití JSON_VALUE pro extrakci jednotlivých hodnot:

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

Extrahování objektů nebo polí

Použití JSON_QUERY pro objekty a pole.

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

Analyzovat pole JSON na řádky

Rozviňte JSON pole do řádků pomocí 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}")

Parsovat objekt JSON na sloupce

Extrahujte jednotlivá pole z objektů JSON pomocí JSON_VALUE() a převádějte výsledky na příslušné typy 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
""")

Upravit JSON data

Použijte JSON_MODIFY k aktualizaci konkrétní cesty v JSON dokumentu bez přepsání celé hodnoty.

Aktualizovat hodnotu JSON

Upravte jednu vlastnost JSON pomocí 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()

Přidat vlastnost JSON

Vložte novou vlastnost do existujícího objektu 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})

Odstraňte vlastnost JSON

Smažte vlastnost z objektu JSON nastavením na 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})

Připojit do JSON pole

Přidejte novou hodnotu na konec JSON pole pomocí append směrnice v 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})

Převod relačních dat na JSON

Klauzule FOR JSON na serverové straně převádí výsledky dotazů na JSON řetězec.

PRO JSON AUTO

Generujte JSON z výsledků dotazu:

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

Získejte větší kontrolu nad strukturou 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": "..."}}

Vnořený JSON

Dotazujte se na datové struktury s vnořenými poli a objekty JSON pomocí poddotazů s FOR JSON a vytvářejte hierarchický výstup 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

Integrační vzory Python

Tyto vzory ukazují, jak vytvářet Python abstrakce nad tabulkami podloženými JSONem.

Vzor repozitáře s JSON

Implementovat vrstvu přístupu k datům, která serializuje a deserializuje Python objekty do sloupců JSON, čímž poskytuje typově bezpečné rozhraní databáze.

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

Připojte se k databázi, poté vytvořte repozitář a použijte ho k uložení a načtení produktu. Metoda save() použije větev INSERT, když má id hodnotu None, a jinak větev 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()

Repozitář vytváří #JsonRepo jako místní dočasnou tabulku vázanou na připojení, které předáte, takže save() a get() musí sdílet stejné připojení. Tabulka se odstraní při uzavření spojení.

Efektivní zpracování velkých JSON výsledků

Když jsou výsledky JSON velké, načítáte je po částech přes více řádků.

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

Převod výsledků dotazů do JSON v Python

Transformujte výsledky relačních dotazů do formátu JSON v Python převodem každého řádku do slovníku a následnou serializací do 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)

Indexování dat JSON

Vytvořte vypočtený sloupec podložený JSON výrazem cesty, aby byla cesta indexovatelná.

Vypočtený sloupec s indexem

Definujte vypočítaný sloupec, který extrahuje hodnotu JSON a aplikuje na ni index pro efektivní filtrování často dotazovaných cest JSON. Následující příklad vytváří trvalou tabulku, přidává trvalý vypočtený sloupec přes cestu JSON $.specs.weight a vytvoří na ní index.

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

Dotaz pomocí indexovaného vypočteného sloupce

Filtrujte přímo podle vypočítaného sloupce. Dotazovací engine používá index místo toho, aby skenoval a analyzoval každý JSON dokument.

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

Vyberte si mezi relačními sloupci a úložištěm JSON

Relační sloupce používejte tehdy, když mají data pevné schéma, potřebují referenční integritu, účastní se JOIN nebo se často objevují v klauzulích WHERE. Používejte JSON sloupce (nvarchar(max)), když jsou data řídká, liší se mezi řádky nebo představují flexibilní konfiguraci či metadata.

Kdy používat serverové a klientské zpracování JSON

Používejte Microsoft SQL JSON funkce (JSON_VALUE, JSON_QUERY, ), OPENJSONkdyž potřebujete filtrovat, indexovat nebo agregovat napříč JSON poli, aniž byste klientovi načítali každý řádek. Tato volba je správná, když pouze podmnožina řádků odpovídá vašim kritériím, nebo když chcete vypočtené sloupcové indexy na JSON cestách.

Používejte klientské Python zpracování (json.loads()), když získáváte celé dokumenty a zpracováváte je v aplikační logice. Tento přístup funguje dobře, když potřebujete celý dokument a nefiltrujete podle JSON polí v databázi.

Pracovní postupy ve stylu dokumentů

Když vaše aplikace ukládá a načítá celé dokumenty, použijte Python-side serializaci a JSON sloupec považujte za neprůhledné úložiště. Zpracovávat a dotazovat dokumenty v Python načítáním a deserializací kompletních JSON blobů:

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

Serverové JSON dotazy

Používejte funkce Microsoft SQL JSON, když potřebujete filtrovat, indexovat nebo agregovat napříč JSON poli, aniž byste museli načítat každý řádek. Tento přístup je efektivnější než načítání všech řádků do Python pro filtrování v paměti:

• JSON_VALUE extrahuje skalární hodnoty a může zpětně vypočítat sloupcové indexy.
• JSON_QUERY extrahuje objekty a pole.
• OPENJSON parsuje JSON do řádků pro operace JOIN a agregaci.
• JSON_MODIFY aktualizuje konkrétní cesty, aniž by přepsal celý dokument.

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

Pro často dotazované JSON cesty vytvořte vypočítaný sloupec s indexem:

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

CREATE INDEX IX_Settings_Department ON Settings(Department);

Osvědčené postupy

Použijte tyto pokyny pro spolehlivé používání JSON sloupců.

Ověřte JSON před uložením

Před uložením ověřte identifikátory JSON a tabulek/sloupců, abyste zabránili injekčním útokům.

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

Nepoužívejte JSON příliš často

Používejte sloupce JSON pro flexibilní nebo řídká data, jako jsou uživatelské preference nebo vlastní pole. Použijte relační sloupce pro:

• Často dotazovaná data.
• Data, která potřebují referenční integritu.
• Sloupce používané ve WHERE větách.

Správně zpracovat None/NULL

Vyřešte chybějící nebo volitelné JSON pole vložením NULL hodnot pro sloupce, které neobsahují data.

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

Související obsah

• Mapování datových typů
• Zpracování řetězců a Unicode
• Dokumentace Microsoft SQL JSON