Manipuler les bases de donnees avec Python : Guide complet des requetes SQL

Introduction

La manipulation des bases de donnees est une competence fondamentale pour tout developpeur Python. Que vous developpiez une application web, un script d’automatisation ou un pipeline de donnees, vous aurez inevitablement besoin d’interagir avec une base de donnees.

Python offre un ecosysteme riche de bibliotheques pour travailler avec differents systemes de gestion de bases de donnees (SGBD). De SQLite pour le prototypage rapide a PostgreSQL pour les applications en production, en passant par MySQL et Oracle, Python s’adapte a tous les besoins.

Dans ce guide complet, nous explorerons les differentes approches pour manipuler les bases de donnees de maniere efficace et securisee. Nous verrons comment eviter les pieges courants comme les injections SQL et comment structurer votre code pour une maintenance optimale.

Pourquoi maitriser les bases de donnees en Python ?

Les bases de donnees sont au coeur de la plupart des applications modernes. Elles permettent de :

• Persister les donnees de maniere fiable et structuree
• Gerer la concurrence avec plusieurs utilisateurs simultanement
• Optimiser les performances grace aux index et au cache
• Garantir l’integrite des donnees avec les contraintes et transactions

Python, avec sa syntaxe claire et ses bibliotheques matures, facilite grandement ces interactions.

Se connecter a la base de donnees

Connexion SQLite (ideal pour le developpement)

SQLite est integre a Python et ne necessite aucune installation supplementaire. C’est parfait pour le prototypage et les tests.

import sqlite3
from contextlib import contextmanager

# Gestionnaire de contexte pour une gestion propre des connexions
@contextmanager
def get_db_connection(db_path: str):
    """Cree une connexion a la base de donnees avec fermeture automatique."""
    conn = sqlite3.connect(db_path)
    conn.row_factory = sqlite3.Row  # Permet l'acces par nom de colonne
    try:
        yield conn
    finally:
        conn.close()

# Utilisation
with get_db_connection('mon_application.db') as conn:
    cursor = conn.cursor()
    cursor.execute('SELECT * FROM utilisateurs')
    for row in cursor.fetchall():
        print(f"Utilisateur: {row['nom']}, Email: {row['email']}")

Connexion PostgreSQL (recommande pour la production)

Pour les applications en production, PostgreSQL offre des fonctionnalites avancees et une meilleure gestion de la concurrence.

import psycopg2
from psycopg2.extras import RealDictCursor

# Configuration de la connexion
config = {
    'host': 'localhost',
    'database': 'ma_base',
    'user': 'mon_utilisateur',
    'password': 'mon_mot_de_passe',
    'port': 5432
}

# Connexion avec curseur dictionnaire
def get_postgres_connection():
    """Etablit une connexion PostgreSQL avec gestion d'erreurs."""
    try:
        conn = psycopg2.connect(**config)
        return conn
    except psycopg2.OperationalError as e:
        print(f"Erreur de connexion: {e}")
        raise

# Utilisation avec context manager
with get_postgres_connection() as conn:
    with conn.cursor(cursor_factory=RealDictCursor) as cur:
        cur.execute("SELECT version()")
        print(cur.fetchone())

Connexion avec pool de connexions

Pour les applications a fort trafic, utilisez un pool de connexions pour optimiser les performances.

from psycopg2 import pool

# Creer un pool de connexions
connection_pool = pool.ThreadedConnectionPool(
    minconn=2,      # Minimum de connexions
    maxconn=10,     # Maximum de connexions
    host='localhost',
    database='ma_base',
    user='utilisateur',
    password='secret'
)

def execute_query(query: str, params: tuple = None):
    """Execute une requete avec une connexion du pool."""
    conn = connection_pool.getconn()
    try:
        with conn.cursor() as cur:
            cur.execute(query, params)
            if query.strip().upper().startswith('SELECT'):
                return cur.fetchall()
            conn.commit()
    finally:
        connection_pool.putconn(conn)

Executer des requetes de maniere securisee

Requetes SELECT avec bind variables

Les bind variables (ou requetes parametrees) sont essentielles pour prevenir les injections SQL.

import sqlite3

def rechercher_utilisateur(conn, email: str, statut: str = 'actif'):
    """Recherche un utilisateur par email de maniere securisee."""
    cursor = conn.cursor()

    # CORRECT: Utilisation de parametres
    query = """
        SELECT id, nom, email, date_creation
        FROM utilisateurs
        WHERE email = ? AND statut = ?
        ORDER BY date_creation DESC
    """
    cursor.execute(query, (email, statut))
    return cursor.fetchone()

def rechercher_par_criteres(conn, criteres: dict):
    """Recherche dynamique avec plusieurs criteres."""
    conditions = []
    params = []

    for colonne, valeur in criteres.items():
        conditions.append(f"{colonne} = ?")
        params.append(valeur)

    query = f"""
        SELECT * FROM utilisateurs
        WHERE {' AND '.join(conditions)}
    """
    cursor = conn.cursor()
    cursor.execute(query, tuple(params))
    return cursor.fetchall()

Requetes INSERT, UPDATE et DELETE

def creer_utilisateur(conn, nom: str, email: str, role: str = 'user'):
    """Cree un nouvel utilisateur et retourne son ID."""
    cursor = conn.cursor()
    query = """
        INSERT INTO utilisateurs (nom, email, role, date_creation)
        VALUES (?, ?, ?, datetime('now'))
    """
    cursor.execute(query, (nom, email, role))
    conn.commit()
    return cursor.lastrowid

def mettre_a_jour_utilisateur(conn, user_id: int, **champs):
    """Met a jour les champs specifies d'un utilisateur."""
    if not champs:
        return False

    set_clause = ', '.join(f"{k} = ?" for k in champs.keys())
    query = f"UPDATE utilisateurs SET {set_clause} WHERE id = ?"

    cursor = conn.cursor()
    cursor.execute(query, (*champs.values(), user_id))
    conn.commit()
    return cursor.rowcount > 0

def supprimer_utilisateur(conn, user_id: int):
    """Supprime un utilisateur (soft delete recommande)."""
    cursor = conn.cursor()
    # Soft delete: marquer comme supprime plutot que supprimer
    query = "UPDATE utilisateurs SET statut = 'supprime', date_suppression = datetime('now') WHERE id = ?"
    cursor.execute(query, (user_id,))
    conn.commit()
    return cursor.rowcount > 0

Insertions en lot avec executemany

Pour inserer plusieurs enregistrements efficacement, utilisez executemany.

def inserer_utilisateurs_en_lot(conn, utilisateurs: list):
    """Insere plusieurs utilisateurs en une seule transaction."""
    cursor = conn.cursor()
    query = """
        INSERT INTO utilisateurs (nom, email, role)
        VALUES (?, ?, ?)
    """

    # Liste de tuples avec les donnees
    donnees = [(u['nom'], u['email'], u.get('role', 'user')) for u in utilisateurs]

    try:
        cursor.executemany(query, donnees)
        conn.commit()
        return len(donnees)
    except Exception as e:
        conn.rollback()
        raise e

Gestion des transactions

Les transactions garantissent l’integrite des donnees lors d’operations multiples.

def transferer_credits(conn, source_id: int, dest_id: int, montant: float):
    """Transfere des credits entre deux utilisateurs de maniere atomique."""
    cursor = conn.cursor()

    try:
        # Verifier le solde source
        cursor.execute("SELECT credits FROM comptes WHERE user_id = ?", (source_id,))
        solde = cursor.fetchone()[0]

        if solde < montant:
            raise ValueError("Solde insuffisant")

        # Debiter le compte source
        cursor.execute(
            "UPDATE comptes SET credits = credits - ? WHERE user_id = ?",
            (montant, source_id)
        )

        # Crediter le compte destination
        cursor.execute(
            "UPDATE comptes SET credits = credits + ? WHERE user_id = ?",
            (montant, dest_id)
        )

        # Enregistrer la transaction
        cursor.execute(
            "INSERT INTO transactions (source, destination, montant, date) VALUES (?, ?, ?, datetime('now'))",
            (source_id, dest_id, montant)
        )

        conn.commit()
        return True

    except Exception as e:
        conn.rollback()  # Annuler toutes les modifications
        print(f"Transaction echouee: {e}")
        raise

Bonnes Pratiques

1. Toujours utiliser les requetes parametrees

Ne concatenez jamais les valeurs utilisateur directement dans vos requetes SQL.

# MAUVAIS - Vulnerable aux injections SQL
query = f"SELECT * FROM users WHERE email = '{email}'"

# BON - Securise avec bind variables
query = "SELECT * FROM users WHERE email = ?"
cursor.execute(query, (email,))

2. Fermer les connexions proprement

Utilisez les context managers pour garantir la fermeture des ressources.

# BON - Fermeture automatique garantie
with sqlite3.connect('db.sqlite') as conn:
    cursor = conn.cursor()
    # ... operations ...
# La connexion est fermee automatiquement

3. Gerer les exceptions de maniere explicite

import sqlite3

def operation_securisee(conn, query, params):
    """Execute une requete avec gestion complete des erreurs."""
    try:
        cursor = conn.cursor()
        cursor.execute(query, params)
        conn.commit()
        return cursor.lastrowid
    except sqlite3.IntegrityError as e:
        conn.rollback()
        print(f"Violation de contrainte: {e}")
        return None
    except sqlite3.OperationalError as e:
        conn.rollback()
        print(f"Erreur operationnelle: {e}")
        raise

4. Utiliser des timeouts de connexion

# SQLite avec timeout
conn = sqlite3.connect('db.sqlite', timeout=10.0)

# PostgreSQL avec timeout
conn = psycopg2.connect(
    host='localhost',
    database='ma_base',
    connect_timeout=10
)

5. Indexer les colonnes frequemment recherchees

def creer_index_performance(conn):
    """Cree les index necessaires pour optimiser les requetes."""
    cursor = conn.cursor()

    # Index sur les colonnes de recherche frequentes
    cursor.execute("CREATE INDEX IF NOT EXISTS idx_users_email ON utilisateurs(email)")
    cursor.execute("CREATE INDEX IF NOT EXISTS idx_users_statut ON utilisateurs(statut)")

    # Index composite pour les requetes combinees
    cursor.execute("CREATE INDEX IF NOT EXISTS idx_users_role_statut ON utilisateurs(role, statut)")

    conn.commit()

Pieges Courants

1. Injection SQL par concatenation de chaines

# DANGER - Ne jamais faire cela !
def recherche_vulnerable(conn, terme):
    query = f"SELECT * FROM produits WHERE nom LIKE '%{terme}%'"
    # Un attaquant peut injecter: "'; DROP TABLE produits; --"

# SOLUTION - Utiliser les parametres
def recherche_securisee(conn, terme):
    query = "SELECT * FROM produits WHERE nom LIKE ?"
    cursor = conn.cursor()
    cursor.execute(query, (f'%{terme}%',))
    return cursor.fetchall()

2. Oublier de commit les modifications

# ERREUR - Les donnees ne sont pas sauvegardees
cursor.execute("INSERT INTO logs (message) VALUES (?)", ("Action effectuee",))
# Manque conn.commit() !

# CORRECT
cursor.execute("INSERT INTO logs (message) VALUES (?)", ("Action effectuee",))
conn.commit()

3. Ne pas gerer les connexions fermees

# ERREUR - La connexion peut etre fermee
def mauvaise_fonction():
    conn = sqlite3.connect('db.sqlite')
    cursor = conn.cursor()
    # Si une exception survient ici, la connexion reste ouverte
    cursor.execute("SELECT * FROM users")
    conn.close()
    return cursor.fetchall()  # Erreur: connexion fermee !

# CORRECT - Recuperer les donnees avant de fermer
def bonne_fonction():
    with sqlite3.connect('db.sqlite') as conn:
        cursor = conn.cursor()
        cursor.execute("SELECT * FROM users")
        resultats = cursor.fetchall()
    return resultats

4. Ignorer les limites de requetes

# DANGER - Peut retourner des millions de lignes
cursor.execute("SELECT * FROM logs")  # Table avec 10M de lignes !

# CORRECT - Toujours limiter et paginer
def get_logs_pagines(conn, page: int = 1, par_page: int = 100):
    offset = (page - 1) * par_page
    cursor = conn.cursor()
    cursor.execute(
        "SELECT * FROM logs ORDER BY date DESC LIMIT ? OFFSET ?",
        (par_page, offset)
    )
    return cursor.fetchall()

Conclusion

La manipulation des bases de donnees en Python est une competence essentielle qui necessite une attention particuliere a la securite et aux performances. En suivant les bonnes pratiques presentees dans ce guide, vous pourrez :

• Ecrire du code securise en utilisant systematiquement les requetes parametrees
• Optimiser les performances avec les pools de connexions et les index
• Garantir l’integrite des donnees grace aux transactions
• Maintenir un code propre avec les context managers et la gestion d’erreurs

N’oubliez pas que la securite doit etre votre priorite numero un. Une seule faille d’injection SQL peut compromettre l’ensemble de votre base de donnees et les donnees de vos utilisateurs.

Pour aller plus loin

• SQLAlchemy : ORM Python pour une abstraction plus elevee
• Alembic : Gestion des migrations de schema
• asyncpg : Driver PostgreSQL asynchrone pour les applications haute performance
• Redis : Cache en memoire pour optimiser les requetes frequentes