Comment maîtriser SQLite avancé en 2026

import sqlite3

conn = sqlite3.connect('blog.db')
cursor = conn.cursor()

# Tables avec relations
cursor.execute('''
CREATE TABLE IF NOT EXISTS users (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    name TEXT NOT NULL,
    email TEXT UNIQUE NOT NULL
);
''')

cursor.execute('''
CREATE TABLE IF NOT EXISTS posts (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    title TEXT NOT NULL,
    content TEXT,
    user_id INTEGER NOT NULL,
    created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
    FOREIGN KEY (user_id) REFERENCES users(id) ON DELETE CASCADE
);
''')

cursor.execute('''
CREATE TABLE IF NOT EXISTS comments (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    content TEXT NOT NULL,
    post_id INTEGER NOT NULL,
    parent_id INTEGER,
    user_id INTEGER NOT NULL,
    created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
    FOREIGN KEY (post_id) REFERENCES posts(id) ON DELETE CASCADE,
    FOREIGN KEY (parent_id) REFERENCES comments(id) ON DELETE CASCADE,
    FOREIGN KEY (user_id) REFERENCES users(id)
);
''')

# Indexes composites pour perf
cursor.execute('CREATE INDEX IF NOT EXISTS idx_posts_user_created ON posts(user_id, created_at);')
cursor.execute('CREATE INDEX IF NOT EXISTS idx_comments_post_parent ON comments(post_id, parent_id);')

conn.commit()
print('Schéma créé avec succès. Fichier: blog.db')
conn.close()

import sqlite3
from datetime import datetime

conn = sqlite3.connect('blog.db')
cursor = conn.cursor()

# Transaction batchée pour 1000 users
users_data = [(f'User{i}', f'user{i}@example.com') for i in range(1, 1001)]
cursor.executemany('INSERT OR IGNORE INTO users (name, email) VALUES (?, ?)', users_data)

# 5000 posts
posts_data = [(f'Post {i}', f'Contenu post {i}', i % 1000 + 1, datetime.now().isoformat()) for i in range(5000)]
cursor.executemany('INSERT INTO posts (title, content, user_id, created_at) VALUES (?, ?, ?, ?)', posts_data)

# 50k commentaires hiérarchiques (batch)
comments_data = []
for i in range(50000):
    post_id = (i % 5000) + 1
    parent_id = i % 100 if i % 10 == 0 else None  # 10% ont parent
    user_id = (i % 1000) + 1
    comments_data.append((f'Comment {i}', post_id, parent_id, user_id, datetime.now().isoformat()))

cursor.executemany('''
INSERT INTO comments (content, post_id, parent_id, user_id, created_at)
VALUES (?, ?, ?, ?, ?)
''', comments_data)

conn.commit()
print(f'Données insérées: {cursor.total_changes} changements')
conn.close()

import sqlite3

conn = sqlite3.connect('blog.db')
cursor = conn.cursor()

cursor.execute('''
WITH RECURSIVE comment_threads AS (
    -- Ancre: commentaires racines
    SELECT id, content, parent_id, 0 as level, created_at
    FROM comments
    WHERE parent_id IS NULL AND post_id = 1

    UNION ALL

    -- Récursion: enfants
    SELECT c.id, c.content, c.parent_id, ct.level + 1, c.created_at
    FROM comments c
    JOIN comment_threads ct ON c.parent_id = ct.id
    WHERE ct.level < 5  -- Limite profondeur
)
SELECT post_id, level, content, created_at FROM comment_threads
ORDER BY level, created_at;
''')

threads = cursor.fetchall()
print(f'Threads pour post 1: {len(threads)} commentaires')
for row in threads[:10]:
    print(row)
conn.close()

import sqlite3

conn = sqlite3.connect('blog.db')
cursor = conn.cursor()

cursor.execute('''
SELECT
    u.name,
    p.title,
    p.created_at,
    COUNT(c.id) OVER (PARTITION BY p.user_id ORDER BY p.created_at ROWS UNBOUNDED PRECEDING) as cumulative_comments,
    ROW_NUMBER() OVER (PARTITION BY p.user_id ORDER BY COUNT(c.id) DESC) as rank_by_comments,
    AVG(LENGTH(p.content)) OVER (PARTITION BY u.id) as avg_post_length
FROM users u
JOIN posts p ON u.id = p.user_id
LEFT JOIN comments c ON p.id = c.post_id
GROUP BY p.id
HAVING rank_by_comments <= 3
ORDER BY u.name, rank_by_comments;
''')

top_posts = cursor.fetchall()
print('Top 3 posts par user par nb commentaires:')
for row in top_posts[:15]:
    print(row)
conn.close()

import sqlite3

conn = sqlite3.connect('blog.db')
cursor = conn.cursor()

# Créer table FTS5 virtuelle
cursor.execute('''
CREATE VIRTUAL TABLE IF NOT EXISTS posts_fts USING fts5(
    title, content, content=posts, content_rowid=posts.id,
    tokenize=porter  -- Stemming anglais, ou 'unicode61' pour FR
);
''')

# Populate FTS (incrémental)
cursor.execute('INSERT INTO posts_fts(rowid, title, content) SELECT id, title, content FROM posts;')

# Query full-text avec ranking
cursor.execute('''
SELECT
    p.title,
    snippet(posts_fts, 1, '<b>', '</b>', '...', 64) as highlight,
    bm25(posts_fts) as score
FROM posts_fts
JOIN posts p ON p.id = posts_fts.rowid
WHERE posts_fts MATCH 'performance OR optim*'
ORDER BY rank
LIMIT 10;
''')

results = cursor.fetchall()
print('Résultats FTS:')
for row in results:
    print(row)
conn.commit()
conn.close()

import sqlite3

conn = sqlite3.connect('blog.db')
cursor = conn.cursor()

# Trigger audit
cursor.execute('''
CREATE TRIGGER IF NOT EXISTS audit_posts
AFTER UPDATE ON posts
FOR EACH ROW
BEGIN
    INSERT INTO posts (title, content, user_id, created_at)
    VALUES (old.title || ' [EDIT]', old.content || ' [EDIT]', old.user_id, CURRENT_TIMESTAMP);
END;
''')

# PRAGMA pour perf WAL + query planner
cursor.executescript('''
PRAGMA journal_mode = WAL;
PRAGMA synchronous = NORMAL;
PRAGMA cache_size = 10000;
PRAGMA optimize;
PRAGMA vacuum;
''')

print('Triggers/PRAGMA appliqués. WAL activé pour concurrency.')

# Test trigger
cursor.execute('UPDATE posts SET title = "Modifié" WHERE id=1;')
conn.commit()
cursor.execute('SELECT * FROM posts WHERE title LIKE "%[EDIT]%";')
print('Audit:', cursor.fetchone())
conn.close()