Bibliothèques de manipulation de bases de données en Python : guide complet

En Python, interagir avec des bases de données est une tâche courante en développement web, analyse de données, auotmatisation ou services backend. L'écosystème propose de nombreuses bibliothèques adaptées à différents types de bases (relationnelles et non relationnelles). Cet article détaille les plus utilisées, avec des exemples concrets et des bonnes pratiques.

Logique fondamentale de l'interaction Python – base de données

Quelle que soit la base ou la bibliothèque, l'interaction suit généralement quatre étapes :

  1. Connexion : paramètres (hôte, port, utilisateur, mot de passe, nom de la base) via l'API de la bibliothèque.
  2. Création d'un objet d'exécution : curseur (cursor) ou session pour exécuter les commandes.
  3. Exécution : requêtes (SELECT, INSERT, UPDATE, DELETE) ou opérations NoSQL.
  4. Fermeture des ressources : d'abord l'objet d'exécution, puis la connexion, dans cet ordre.

En production, privilégiez les pools de connexions et gérez les exceptions pour éviter les plantages.

Bibliothèques pour bases relationnelles

1. PyMySQL – MySQL

Présentation : client MySQL pur Python, compatible Python 3, multiplateforme. Installez avec pip install pymysql.

Exemple complet (création de table, insertion, requête, mise à jour, suppression) :

import pymysql
from pymysql import Error

def demo_mysql():
    config = {
        "host": "localhost",
        "port": 3306,
        "user": "root",
        "password": "123456",
        "database": "test_db"
    }
    conn = None
    cur = None
    try:
        conn = pymysql.connect(**config)
        cur = conn.cursor()
        # Création table
        cur.execute("""
            CREATE TABLE IF NOT EXISTS members (
                id INT AUTO_INCREMENT PRIMARY KEY,
                full_name VARCHAR(100) NOT NULL,
                age INT,
                city VARCHAR(50)
            ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4
        """)
        # Insertion unique
        cur.execute("INSERT INTO members (full_name, age, city) VALUES (%s, %s, %s)",
                    ("Alice", 30, "Paris"))
        conn.commit()
        # Insertion multiple
        data = [("Bob", 25, "Lyon"), ("Charlie", 35, "Marseille")]
        cur.executemany("INSERT INTO members (full_name, age, city) VALUES (%s, %s, %s)", data)
        conn.commit()
        # Requête
        cur.execute("SELECT * FROM members WHERE age > %s", (28,))
        results = cur.fetchall()
        for row in results:
            print(f"ID: {row[0]}, Nom: {row[1]}, Âge: {row[2]}, Ville: {row[3]}")
        # Mise à jour
        cur.execute("UPDATE members SET age = %s WHERE full_name = %s", (31, "Alice"))
        conn.commit()
        # Suppression
        cur.execute("DELETE FROM members WHERE full_name = %s", ("Charlie",))
        conn.commit()
    except Error as e:
        print(f"Erreur : {e}")
        if conn:
            conn.rollback()
    finally:
        if cur:
            cur.close()
        if conn and conn.open:
            conn.close()

if __name__ == "__main__":
    demo_mysql()

Bonnes pratiques : utilisez %s pour éviter les injections SQL, gérez les transactions (commit / rollback), fermez les ressources dans finally, et en production, employez un pool (ex. DBUtils).

2. sqlite3 – SQLite (intégré)

Présentation : bibliothèque standard de Python, aucun安装 nécessaire. Convient aux projets légers, tests ou embarqués. Fichier unique.

import sqlite3
from sqlite3 import Error

def demo_sqlite():
    chemin = "exemple.db"
    conn = None
    cur = None
    try:
        conn = sqlite3.connect(chemin)
        cur = conn.cursor()
        cur.execute("""
            CREATE TABLE IF NOT EXISTS livres (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                titre TEXT NOT NULL,
                auteur TEXT,
                prix REAL
            )
        """)
        cur.execute("INSERT INTO livres (titre, auteur, prix) VALUES (?, ?, ?)",
                    ("Python pour tous", "Dupont", 24.99))
        cur.executemany("INSERT INTO livres (titre, auteur, prix) VALUES (?, ?, ?)",
                        [("Java avancé", "Martin", 34.50), ("SQL facile", "Legrand", 19.90)])
        conn.commit()
        cur.execute("SELECT * FROM livres WHERE prix > ?", (20,))
        for row in cur.fetchall():
            print(row)
        cur.execute("UPDATE livres SET prix = ? WHERE titre = ?", (22.99, "Python pour tous"))
        conn.commit()
        cur.execute("DELETE FROM livres WHERE auteur = ?", ("Martin",))
        conn.commit()
    except Error as e:
        print(e)
        if conn:
            conn.rollback()
    finally:
        if cur:
            cur.close()
        if conn:
            conn.close()

if __name__ == "__main__":
    demo_sqlite()

Remarques : utilisez ? comme placeholders (pas %s), l'auto-incrémentation se fait avec INTEGER PRIMARY KEY AUTOINCREMENT, et SQLite ne gère pas bien les écritures simultanées.

3. psycopg2 – PostgreSQL

Présentation : client officiel PostgreSQL, performant. Installez psycopg2-binary pour éviter la compilation. Port par défaut : 5432.

import psycopg2
from psycopg2 import Error

def demo_postgres():
    config = {
        "host": "localhost",
        "port": 5432,
        "user": "postgres",
        "password": "123456",
        "database": "test_pg"
    }
    conn = None
    cur = None
    try:
        conn = psycopg2.connect(**config)
        cur = conn.cursor()
        cur.execute("""
            CREATE TABLE IF NOT EXISTS employes (
                id SERIAL PRIMARY KEY,
                nom VARCHAR(50) NOT NULL,
                salaire NUMERIC,
                extra JSONB
            )
        """)
        cur.execute("INSERT INTO employes (nom, salaire, extra) VALUES (%s, %s, %s)",
                    ("Jean", 45000, '{"service": "IT"}'))
        cur.executemany("INSERT INTO employes (nom, salaire, extra) VALUES (%s, %s, %s)",
                        [("Marie", 52000, '{"service": "RH"}'), ("Pierre", 48000, '{"service": "IT"}')])
        conn.commit()
        cur.execute("SELECT * FROM employes WHERE extra->>'service' = %s", ("IT",))
        for row in cur.fetchall():
            print(f"ID: {row[0]}, Nom: {row[1]}, Salaire: {row[2]}, Extra: {row[3]}")
        cur.execute("UPDATE employes SET salaire = %s WHERE nom = %s", (47000, "Jean"))
        conn.commit()
    except Error as e:
        print(e)
        if conn:
            conn.rollback()
    finally:
        if cur:
            cur.close()
        if conn:
            conn.close()

if __name__ == "__main__":
    demo_postgres()

Spécificités : auto-incrémentation avec SERIAL, support natif du JSON/JSONB, placeholders %s comme MySQL.

Bibliothèques pour bases NoSQL

4. pymongo – MongoDB

MongoDB stocke les données sous forme de documents JSON (BSON). Pas de schéma fixe. Installez pip install pymongo.

from pymongo import MongoClient
from pymongo.errors import PyMongoError

def demo_mongo():
    client = MongoClient("mongodb://localhost:27017/")
    db = client["demo_db"]
    collection = db["utilisateurs"]
    try:
        # Insertion unique
        doc1 = {"nom": "Alice", "age": 28, "adresse": {"ville": "Paris", "code": 75001}}
        result = collection.insert_one(doc1)
        print(f"ID inséré : {result.inserted_id}")
        # Insertion multiple
        docs = [
            {"nom": "Bob", "age": 32, "adresse": {"ville": "Lyon", "code": 69001}},
            {"nom": "Charlie", "age": 24, "adresse": {"ville": "Marseille", "code": 13001}}
        ]
        collection.insert_many(docs)
        # Requête
        for doc in collection.find({"age": {"$gt": 25}}, {"_id": 0}):
            print(doc)
        # Mise à jour
        collection.update_one({"nom": "Alice"}, {"$set": {"age": 29}})
        # Suppression
        collection.delete_one({"nom": "Charlie"})
    except PyMongoError as e:
        print(f"Erreur MongoDB : {e}")
    finally:
        client.close()

if __name__ == "__main__":
    demo_mongo()

Conseils : utilisez toujours des opérateurs comme $set, $inc pour les mises à jour, créez des index pour les performances, et fermez la connexion.

5. redis-py – Redis

Redis est un magasin clé-valeur en mémoire. Installez pip install redis. Ne pas oublier decode_responses=True pour obtenir des chaînes.

import redis

def demo_redis():
    r = redis.Redis(host="localhost", port=6379, decode_responses=True)
    try:
        # String
        r.set("user:1", "Alice", ex=120)
        print(r.get("user:1"))
        # Hash
        r.hset("product:1", mapping={"name": "Laptop", "price": 999})
        print(r.hgetall("product:1"))
        # Liste
        r.rpush("queue", "task1", "task2")
        print(r.lrange("queue", 0, -1))
        # Set
        r.sadd("tags", "python", "redis", "tutorial")
        print(r.smembers("tags"))
    finally:
        r.close()

if __name__ == "__main__":
    demo_redis()

Important : utilisez un pool de connexions en production, définissez des TTL pour éviter les fuites mémoire, et sérialisez les données complexes avec json.

Bonnes pratiques générales

  • Gestion des exceptions : capturez toutes les erreurs et effectuez un rollback si nécessaire.
  • Pool de connexions : pour MySQL (DBUtils), MongoDB (intégré), Redis (ConnectionPool).
  • Injection SQL : toujours utiliser des paramètres (placeholders).
  • Journalisation : logguez les événements clés.
  • Sécurité : mots de passe, restriction IP, variables d'environnement.
  • Opérations par lots : privilégiez les méthodes executemany, insert_many.

Récapitulatif du choix des bbiliothèques

  • MySQL : PyMySQL
  • SQLite : sqlite3 (intégré)
  • PostgreSQL : psycopg2-binary
  • MongoDB : pymongo (ou motor pour asynchrone)
  • Redis : redis-py (ou aioredis pour asynchrone)

Ces exemples sont prêts à être exécutés après installation des dépendances. Adaptez les paramètres de connexion à votre environnement.

Étiquettes: PyMySQL sqlite3 psycopg2 pymongo redis-py

Publié le 6 juin à 21h39

Étiquettes Populaires

©2026 L'Atelier Monstre