Gebruik JSON-data met mssql-python

Microsoft SQL 2016 en latere versies en Azure SQL bieden JSON-ondersteuning via functies die op nvarchar kolommen werken. De mssql-python-driver verzendt en ontvangt JSON als gewone Python-strings. U kunt:

  • Sla JSON op als strings in nvarchar kolommen.
  • Zoek JSON op met padexpressies met behulp van JSON_VALUE, JSON_QUERY, en OPENJSON.
  • Transformeer relationele data naar JSON met FOR JSON.
  • Parsen JSON in relationeel formaat met OPENJSON.

Opmerking

Microsoft SQL slaat JSON-gegevens op in kolommen, niet in nvarchar een speciaal JSON-kolomtype. De mssql-python-driver verzendt en ontvangt JSON als reguliere strings. Gebruik de ingebouwde json module van Python om aan de clientzijde te serialiseren en deserialiseren.

Sla JSON-gegevens op

Serialiseer Python-dicts naar strings met json.dumps() voordat je ze invoegt in nvarchar-kolommen.

JSON-tekenreeks invoegen

Sla een Python-woordenboek op als JSON-tekst in een databasetabel:

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

Valideer JSON bij het invoegen

Gebruik de ISJSON() functie om de JSON-syntaxis te valideren voordat je het volgende invoegt:

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

Query's uitvoeren op JSON-gegevens

Gebruik Microsoft SQL JSON-padfuncties om waarden op de server te extraheren voordat je ze terugstuurt naar de client.

Extraheren van scalairwaarden

Gebruik JSON_VALUE om losse waarden te extraheren:

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

Objecten of matrices extraheren

Gebruik JSON_QUERY voor objecten en 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}")

JSON-array parseren naar rijen

Breid een JSON-array uit tot rijen met behulp van: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-object parseren naar kolommen

Haal individuele velden uit JSON-objecten met behulp JSON_VALUE() en cast de resultaten naar de juiste SQL-typen.

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

Wijzig JSON-gegevens

Gebruik JSON_MODIFY om een specifiek pad in een JSON-document bij te werken zonder de hele waarde te herschrijven.

JSON-waarde bijwerken

Wijzig een enkele JSON-eigenschap met behulp van 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()

Voeg JSON-eigenschap toe

Voeg een nieuwe eigenschap in in een bestaand JSON-object:

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

Verwijder JSON-eigenschap

Verwijder een eigenschap uit een JSON-object door deze in te stellen op 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})

Toevoegen aan JSON-array

Voeg een nieuwe waarde toe aan het einde van een JSON-array met behulp van de richtlijn append in 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})

Relationele gegevens converteren naar JSON

De FOR JSON clausule zet zoekresultaten om in een JSON-string aan de serverzijde.

VOOR JSON AUTO

Genereer JSON uit zoekresultaten:

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

VOOR JSON PATH

Krijg meer controle over de JSON-structuur:

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": "..."}}

Geneste JSON

Zoek datastructuren op met geneste JSON-arrays en objecten door gebruik te maken van subqueries met FOR JSON om hiërarchische JSON-output te construeren.

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

Deze patronen laten zien hoe je Python-abstracties kunt bouwen over JSON-gebackde tabellen.

Repository-patroon met JSON

Implementeer een data-toegangslaag die Python-objecten serialiseert en deserialiseert naar JSON-kolommen, waardoor een typeveilige interface met de database wordt geboden.

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

Maak verbinding met de database, maak vervolgens de repository aan en gebruik deze om een product op te slaan en op te halen. De methode save() volgt de vertakking INSERT als idNone is en anders de vertakking 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()

De repository maakt #JsonRepo aan als een lokale tijdelijke tabel die is gekoppeld aan de verbinding die je doorgeeft, dus save() en get() moeten diezelfde verbinding delen. De tabel wordt verwijderd wanneer de verbinding wordt verbroken.

Efficiënt omgaan met grote JSON-resultaten

Wanneer JSON-resultaten groot zijn, haal ze dan in delen op over meerdere rijen.

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

Converteer zoekresultaten naar JSON in Python

Transformeer relationele queryresultaten in JSON-formaat in Python door elke rij om te zetten naar een woordenboek en vervolgens te serialiseren naar 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-gegevens indexeren

Maak een berekende kolom met een JSON-padexpressie om het pad indexeerbaar te maken.

Berekende kolom met index

Definieer een berekende kolom die een JSON-waarde extrahert en pas daarop een index toe voor efficiënte filtering op vaak geraadpleegde JSON-paden. Het volgende voorbeeld maakt een permanente tabel aan, voegt een behouden berekende kolom toe over het JSON-pad $.specs.weight en maakt een index daarop.

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

Query met behulp van geïndexeerde berekende kolom

Filter direct op de berekende kolom. De query-engine gebruikt de index in plaats van elk JSON-document te scannen en te parsen.

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

Kies tussen relationele kolommen en JSON-opslag

Gebruik relationele kolommen wanneer data een vast schema heeft, referentieintegriteit nodig heeft, deelneemt aan JOINs of vaak voorkomt in WHERE-clausules. Gebruik JSON-kolommen (nvarchar(max)) wanneer data schaars is, varieert tussen rijen, of flexibele configuratie of metadata vertegenwoordigt.

Wanneer gebruik je server-side in plaats van client-side JSON-verwerking

Gebruik Microsoft SQL JSON-functies (JSON_VALUE, JSON_QUERY, OPENJSON) wanneer je moet filteren, indexeren of aggregeren over JSON-velden zonder elke rij naar de client te halen. Deze keuze is juist wanneer slechts een deelverzameling van rijen aan je criteria voldoet, of wanneer je berekende kolomindexen op JSON-paden wilt.

Gebruik client-side Python processing (json.loads()) wanneer je volledige documenten ophaalt en verwerkt in applicatielogica. Deze aanpak werkt goed als je het volledige document nodig hebt en niet filtert op JSON-velden in de database.

Workflows in documentstijl

Wanneer je applicatie volledige documenten opslaat en ophaalt, gebruik dan Python-side serialisatie en behandel de JSON-kolom als ondoorzichtige opslag. Verwerk en raadpleeg de documenten in Python door volledige JSON-blobs op te halen en te deserialiseren:

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-query's aan serverzijde

Gebruik Microsoft SQL JSON-functies wanneer je moet filteren, indexeren of aggregeren over JSON-velden zonder elke rij op te halen. Deze aanpak is efficiënter dan alle rijen in Python laden om in het geheugen te filteren:

  • JSON_VALUE extraheert scalaire waarden en kan dienen als basis voor indexen op berekende kolommen.
  • JSON_QUERY extraheert objecten en arrays.
  • OPENJSON pakt JSON uit naar rijen voor JOIN's en aggregatie.
  • JSON_MODIFY Werkt specifieke paden bij zonder het hele document te herschrijven.
# 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"})

Voor vaak bevraagde JSON-paden, maak een berekende kolom met een index:

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

CREATE INDEX IX_Settings_Department ON Settings(Department);

Beste praktijken

Pas deze richtlijnen toe om JSON-kolommen betrouwbaar te gebruiken.

Valideer JSON vóór opslag

Valideer JSON en tabel-/kolomidentificaties voordat je opslaat om injectieaanvallen te voorkomen.

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

Gebruik JSON niet te veel

Gebruik JSON-kolommen voor flexibele of spaarzame data, zoals gebruikersvoorkeuren of aangepaste velden. Gebruik relationele kolommen voor:

  • Vaak geraadpleegde gegevens.
  • Data die referentieintegriteit nodig heeft.
  • Kolommen die in WHERE-clausules worden gebruikt.

Ga met None/NULL goed om

Behandel ontbrekende of optionele JSON-velden door NULL-waarden in te voegen voor kolommen zonder 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}")