Traitement de JSON avec SQLite

Introduction

Dans ce laboratoire, vous apprendrez à traiter les données JSON dans SQLite. Vous explorerez comment stocker, extraire, filtrer et mettre à jour des données JSON dans une base de données SQLite. Ce laboratoire offre une introduction pratique à l'utilisation des données JSON dans SQLite, une compétence de plus en plus précieuse dans la gestion moderne des données.

Skills Graph

Créer une base de données et une table

Dans cette étape, vous allez créer une base de données SQLite et une table pour stocker des données JSON. SQLite est une base de données légère qui stocke les données dans un seul fichier, ce qui facilite sa gestion.

Tout d'abord, ouvrez votre terminal. Le chemin d'accès par défaut est /home/labex/project.

Maintenant, créons un répertoire pour stocker notre base de données.

mkdir sqlite_json
cd sqlite_json

Ces commandes créent un répertoire nommé sqlite_json, puis modifient le répertoire courant pour y accéder. Cela permettra de garder vos fichiers de projet organisés.

Ensuite, créez une base de données SQLite nommée mydatabase.db.

sqlite3 mydatabase.db

Cette commande ouvre l'interpréteur (shell) SQLite, en se connectant à la base de données mydatabase.db. Si le fichier de base de données n'existe pas, SQLite le créera.

Maintenant, créez une table nommée products avec des colonnes pour id, name et details. La colonne details stockera les données JSON sous forme de texte.

CREATE TABLE products (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    name TEXT,
    details TEXT
);

Cette commande SQL crée la table products :

• id : Un entier unique qui s'incrémente automatiquement pour chaque nouveau produit.
• name : Le nom du produit (par exemple, Laptop, Smartphone).
• details : Un champ texte pour stocker les données JSON du produit.

Insérer des données JSON

Dans cette étape, vous allez insérer des données JSON dans la table products.

Insérons deux exemples d'enregistrements dans la table products.

INSERT INTO products (name, details) VALUES (
    'Laptop',
    '{"brand": "Dell", "model": "XPS 13", "specs": {"cpu": "Intel i7", "memory": "16GB", "storage": "512GB SSD"}}'
);

INSERT INTO products (name, details) VALUES (
    'Smartphone',
    '{"brand": "Samsung", "model": "Galaxy S21", "specs": {"display": "6.2 inch", "camera": "12MP", "storage": "128GB"}}'
);

Ces instructions INSERT ajoutent deux lignes à la table products. La colonne details contient des données JSON sous forme de chaîne de texte.

Pour vérifier que les données ont été insérées correctement, exécutez la requête suivante :

SELECT * FROM products;

Résultat attendu :

1|Laptop|{"brand": "Dell", "model": "XPS 13", "specs": {"cpu": "Intel i7", "memory": "16GB", "storage": "512GB SSD"}}
2|Smartphone|{"brand": "Samsung", "model": "Galaxy S21", "specs": {"display": "6.2 inch", "camera": "12MP", "storage": "128GB"}}

Ce résultat confirme que les données JSON ont été stockées avec succès dans la table products.

Extraire des champs JSON avec une fonction personnalisée

Étant donné que SQLite n'a pas de fonctions JSON intégrées, vous allez créer une fonction Python personnalisée pour extraire des données des chaînes JSON.

Tout d'abord, quittez l'interpréteur SQLite (shell).

.exit

Maintenant, créez un fichier Python nommé json_extractor.py.

nano json_extractor.py

Collez le code Python suivant dans le fichier json_extractor.py :

import sqlite3
import json

def json_extract(json_str, path):
    try:
        json_data = json.loads(json_str)
        path_components = path.split('.')
        value = json_data
        for component in path_components:
            value = value.get(component)
        return value
    except (json.JSONDecodeError, AttributeError, TypeError):
        return None

def connect_db(db_path):
    conn = sqlite3.connect(db_path)
    conn.create_function("json_extract", 2, json_extract)
    return conn

if __name__ == '__main__':
    conn = connect_db('mydatabase.db')
    cursor = conn.cursor()

    cursor.execute("SELECT json_extract(details, 'brand') FROM products WHERE name = 'Laptop'")
    print(cursor.fetchone()[0])

    cursor.execute("SELECT json_extract(details, 'specs.cpu') FROM products WHERE name = 'Laptop'")
    print(cursor.fetchone()[0])

    conn.close()

Ce code Python définit une fonction json_extract qui prend une chaîne JSON et un chemin d'accès (path) en entrée et renvoie la valeur à ce chemin. Il inclut également une fonction connect_db pour se connecter à la base de données SQLite et enregistrer la fonction json_extract.

• json.loads(json_str) : Cette ligne analyse (parse) la chaîne JSON en un dictionnaire Python.
• path.split('.') : Cela divise le chemin en une liste de composants. Par exemple, 'specs.cpu' devient ['specs', 'cpu'].
• La boucle itère à travers les composants du chemin, accédant aux valeurs imbriquées dans les données JSON.

Enregistrez le fichier et quittez nano.

Maintenant, exécutez le script Python.

python3 json_extractor.py

Résultat attendu :

Dell
Intel i7

Ce script se connecte à la base de données, enregistre la fonction json_extract, puis l'utilise pour extraire la marque (brand) et le CPU du Laptop.

Filtrer les données à l'aide de requêtes JSON

Dans cette étape, vous utiliserez la fonction personnalisée json_extract pour filtrer les données en fonction des valeurs contenues dans les champs JSON.

Ouvrez à nouveau le fichier json_extractor.py.

nano json_extractor.py

Modifiez le fichier json_extractor.py pour inclure une fonction permettant d'interroger la base de données :

import sqlite3
import json

def json_extract(json_str, path):
    try:
        json_data = json.loads(json_str)
        path_components = path.split('.')
        value = json_data
        for component in path_components:
            value = value.get(component)
        return value
    except (json.JSONDecodeError, AttributeError, TypeError):
        return None

def connect_db(db_path):
    conn = sqlite3.connect(db_path)
    conn.create_function("json_extract", 2, json_extract)
    return conn

def filter_products(db_path, json_path, value):
    conn = connect_db(db_path)
    cursor = conn.cursor()
    query = f"SELECT * FROM products WHERE json_extract(details, '{json_path}') = '{value}'"
    cursor.execute(query)
    results = cursor.fetchall()
    conn.close()
    return results

if __name__ == '__main__':
    ## Example usage:
    dell_products = filter_products('mydatabase.db', 'brand', 'Dell')
    print("Products with brand 'Dell':", dell_products)

    intel_products = filter_products('mydatabase.db', 'specs.cpu', 'Intel i7')
    print("Products with CPU 'Intel i7':", intel_products)

Ce code ajoute une fonction filter_products qui prend un chemin de base de données (database path), un chemin JSON (JSON path) et une valeur en entrée. Il se connecte ensuite à la base de données, enregistre la fonction json_extract et exécute une requête pour trouver tous les produits où la valeur au chemin JSON spécifié correspond à la valeur donnée.

Enregistrez le fichier et quittez nano.

Maintenant, exécutez le script Python.

python3 json_extractor.py

Résultat attendu :

Products with brand 'Dell': [(1, 'Laptop', '{"brand": "Dell", "model": "XPS 13", "specs": {"cpu": "Intel i7", "memory": "16GB", "storage": "512GB SSD"}}')]
Products with CPU 'Intel i7': [(1, 'Laptop', '{"brand": "Dell", "model": "XPS 13", "specs": {"cpu": "Intel i7", "memory": "16GB", "storage": "512GB SSD"}}')]

Ce résultat montre les produits qui correspondent aux critères spécifiés.

Mettre à jour les valeurs JSON

Dans cette étape, vous apprendrez à mettre à jour les valeurs dans les champs JSON.

Ouvrez le fichier json_extractor.py.

nano json_extractor.py

Modifiez le fichier json_extractor.py pour inclure des fonctions permettant de mettre à jour le JSON et la base de données :

import sqlite3
import json

def json_extract(json_str, path):
    try:
        json_data = json.loads(json_str)
        path_components = path.split('.')
        value = json_data
        for component in path_components:
            value = value.get(component)
        return value
    except (json.JSONDecodeError, AttributeError, TypeError):
        return None

def connect_db(db_path):
    conn = sqlite3.connect(db_path)
    conn.create_function("json_extract", 2, json_extract)
    return conn

def filter_products(db_path, json_path, value):
    conn = connect_db(db_path)
    cursor = conn.cursor()
    query = f"SELECT * FROM products WHERE json_extract(details, '{json_path}') = '{value}'"
    cursor.execute(query)
    results = cursor.fetchall()
    conn.close()
    return results

def update_json_value(json_str, path, new_value):
    try:
        json_data = json.loads(json_str)
        path_components = path.split('.')
        target = json_data
        for i in range(len(path_components) - 1):
            target = target.get(path_components[i])

        target[path_components[-1]] = new_value
        return json.dumps(json_data)
    except (json.JSONDecodeError, AttributeError, TypeError):
        return json_str  ## Return original if update fails

def update_product_details(db_path, product_name, json_path, new_value):
    conn = sqlite3.connect(db_path)
    cursor = conn.cursor()

    ## Get the current details
    cursor.execute("SELECT details FROM products WHERE name = ?", (product_name,))
    result = cursor.fetchone()
    if not result:
        conn.close()
        return False

    current_details = result[0]

    ## Update the JSON
    updated_details = update_json_value(current_details, json_path, new_value)

    ## Update the database
    cursor.execute("UPDATE products SET details = ? WHERE name = ?", (updated_details, product_name))
    conn.commit()
    conn.close()
    return True

if __name__ == '__main__':
    ## Example usage:
    update_product_details('mydatabase.db', 'Laptop', 'specs.memory', '32GB')
    print("Laptop memory updated to 32GB")

    ## Verify the update
    conn = sqlite3.connect('mydatabase.db')
    cursor = conn.cursor()
    cursor.execute("SELECT details FROM products WHERE name = 'Laptop'")
    updated_details = cursor.fetchone()[0]
    print("Updated Laptop details:", updated_details)
    conn.close()

Ce code ajoute deux fonctions :

• update_json_value : Cette fonction prend une chaîne JSON, un chemin (path) et une nouvelle valeur en entrée. Elle analyse (parse) la chaîne JSON, met à jour la valeur au chemin spécifié et renvoie la chaîne JSON mise à jour.
• update_product_details : Cette fonction prend un chemin de base de données (database path), un nom de produit, un chemin JSON (JSON path) et une nouvelle valeur en entrée. Elle se connecte à la base de données, récupère les données JSON actuelles pour le produit, met à jour la valeur au chemin spécifié à l'aide de update_json_value, puis met à jour la base de données avec les données JSON modifiées.

Enregistrez le fichier et quittez nano.

Maintenant, exécutez le script Python.

python3 json_extractor.py

Résultat attendu :

Laptop memory updated to 32GB
Updated Laptop details: {"brand": "Dell", "model": "XPS 13", "specs": {"cpu": "Intel i7", "memory": "32GB", "storage": "512GB SSD"}}

Ce résultat confirme que la mémoire du Laptop a été mise à jour à 32GB dans la base de données.

Résumé

Dans ce labo, vous avez appris à traiter des données JSON dans SQLite. Vous avez commencé par créer une base de données et une table pour stocker des données JSON. Ensuite, vous avez appris à insérer des données JSON dans la table. Vous avez créé une fonction Python personnalisée pour extraire des champs spécifiques des données JSON et vous avez utilisé cette fonction pour filtrer les données en fonction des valeurs contenues dans les champs JSON. Enfin, vous avez appris à mettre à jour les valeurs dans les champs JSON à l'aide d'une fonction Python personnalisée. Ces compétences fournissent une base pour gérer efficacement les données JSON dans les bases de données SQLite.