Accès aux bases de données Python

Accès aux bases de données avec Python (utilisation de MySQL Connector)

Python offre une manière standardisée d’interagir avec diverses bases de données via la spécification de l’API de base de données Python v2.0 (DB-API 2). La plupart des pilotes/connecteurs de base de données pour Python adhèrent à cette norme.

Vous devez installer un pilote spécifique compatible DB-API pour la base de données à laquelle vous souhaitez vous connecter (par exemple, PostgreSQL, SQLite, Oracle, SQL Server, MySQL). Ce tutoriel se concentre sur MySQL en utilisant le pilote officiel mysql-connector-python.

Le workflow du DB-API implique généralement :

Installation de mysql-connector-python

Avant de commencer, installez le pilote en utilisant pip :

pip install mysql-connector-python

Vérifiez l’installation en essayant de l’importer dans un interpréteur Python :

import mysql.connector

Si cela s’exécute sans un ImportError, le pilote est installé.

Prérequis

Pour les exemples ci-dessous, supposons :

Établissement d’une connexion à la base de données

Utilisez mysql.connector.connect() pour établir une connexion. Cela renvoie un objet de connexion.

import mysql.connector
from mysql.connector import Error

def create_connection(host_name, user_name, user_password, db_name):
    """Crée une connexion à la base de données MySQL spécifiée."""
    connection = None
    try:
        connection = mysql.connector.connect(
            host=host_name,
            user=user_name,
            password=user_password,
            database=db_name
        )
        print("Connexion à la base de données MySQL réussie")
    except Error as e:
        print(f"L'erreur '{e}' s'est produite")
    return connection

# Exemple d'utilisation (remplacez par vos détails)
conn = create_connection("localhost", "testuser", "test1234", "testdb")

# N'oubliez pas de fermer la connexion lorsque vous avez terminé
# if conn and conn.is_connected():
#    conn.close()
#    print("La connexion MySQL est fermée")

L’utilisation de gestionnaires de contexte (instruction with) est la méthode recommandée pour gérer les connexions et les curseurs, car ils garantissent la fermeture automatique des ressources.

Création d’une table de base de données

Une fois connecté, créez un objet curseur en utilisant connection.cursor(). Utilisez la méthode execute() du curseur pour exécuter des commandes SQL DDL (Data Definition Language) comme CREATE TABLE.

import mysql.connector
from mysql.connector import Error

# Supposons que la fonction create_connection existe comme défini ci-dessus

def execute_query(connection, query):
    """Exécute une seule requête SQL."""
    cursor = connection.cursor()
    try:
        cursor.execute(query)
        connection.commit() # Important pour DDL/DML si autocommit=False
        print("Requête exécutée avec succès")
    except Error as e:
        print(f"L'erreur '{e}' s'est produite")
    finally:
        if cursor:
             cursor.close()

# -- Exécution principale --
conn = create_connection("localhost", "testuser", "test1234", "testdb")

if conn and conn.is_connected():
    # Supprime la table si elle existe (pour démonstration)
    drop_table_query = "DROP TABLE IF EXISTS employees;"
    execute_query(conn, drop_table_query)

    # Crée la table des employés
    create_employees_table = """
    CREATE TABLE employees (
      id INT AUTO_INCREMENT PRIMARY KEY,
      first_name VARCHAR(50) NOT NULL,
      last_name VARCHAR(50),
      age INT,
      gender CHAR(1),
      salary FLOAT
    );
    """
    execute_query(conn, create_employees_table)

    conn.close()
    print("La connexion MySQL est fermée")

Opération INSERT (Création d’enregistrements)

Utilisez la méthode execute() avec une instruction INSERT. Crucialement, utilisez des requêtes paramétrées (placeholders comme %s) pour prévenir les vulnérabilités d’injection SQL. Passez les valeurs sous forme de tuple ou de liste comme deuxième argument à execute().

# (En continuant l'exemple précédent, en supposant que la connexion 'conn' est ouverte)

def insert_employee(connection, employee_data):
    """Insère un nouvel employé dans la table des employés."""
    sql = """
    INSERT INTO employees (first_name, last_name, age, gender, salary)
    VALUES (%s, %s, %s, %s, %s);
    """
    cursor = connection.cursor()
    try:
        cursor.execute(sql, employee_data)
        connection.commit() # Valide la transaction
        print(f"Employé {employee_data[0]} ajouté avec succès. ID : {cursor.lastrowid}")
        return cursor.lastrowid
    except Error as e:
        print(f"Échec de l'insertion de l'enregistrement : {e}")
        connection.rollback() # Annule en cas d'erreur
    finally:
        if cursor:
            cursor.close()
    return None

# -- À l'intérieur du bloc d'exécution principal --
if conn and conn.is_connected():
    # ... (code de création de table) ...

    # Insère un seul employé
    employee1 = ('Maria', 'Garcia', 30, 'F', 55000.00)
    insert_employee(conn, employee1)

    # Insère plusieurs employés en utilisant executemany
    employees_to_add = [
        ('John', 'Doe', 45, 'M', 75000.00),
        ('Lisa', 'Ray', 28, 'F', 62000.00)
    ]
    sql_insert_many = "INSERT INTO employees (first_name, last_name, age, gender, salary) VALUES (%s, %s, %s, %s, %s);"
    cursor = conn.cursor()
    try:
        cursor.executemany(sql_insert_many, employees_to_add)
        conn.commit()
        print(f"{cursor.rowcount} employés insérés avec succès.")
    except Error as e:
        print(f"Échec de l'insertion de plusieurs enregistrements : {e}")
        conn.rollback()
    finally:
        if cursor:
            cursor.close()

    # ... (fermeture de la connexion) ...

Opération READ (Récupération des données)

Utilisez execute() avec une instruction SELECT. Ensuite, utilisez des méthodes de récupération comme fetchone(), fetchall() ou fetchmany(size) sur le curseur pour récupérer les résultats.

# (En continuant l'exemple précédent, en supposant que la connexion 'conn' est ouverte)

def fetch_all_employees(connection):
    """Récupère tous les enregistrements d'employés."""
    sql = "SELECT id, first_name, last_name, age, salary FROM employees;"
    cursor = connection.cursor(dictionary=True) # Récupère sous forme de dictionnaires
    employees = []
    try:
        cursor.execute(sql)
        employees = cursor.fetchall()
        print(f"{len(employees)} enregistrements d'employés récupérés.")
    except Error as e:
        print(f"Échec de la récupération des enregistrements : {e}")
    finally:
        if cursor:
            cursor.close()
    return employees

# -- À l'intérieur du bloc d'exécution principal --
if conn and conn.is_connected():
    # ... (code de création de table, d'insertion) ...

    all_employees = fetch_all_employees(conn)
    if all_employees:
        for employee in all_employees:
             # Accès par nom de colonne si dictionary=True, sinon par index (par exemple, employee[1])
            print(f"ID : {employee['id']}, Nom : {employee['first_name']} {employee['last_name']}, Âge : {employee['age']}")

    # ... (fermeture de la connexion) ...

Définir dictionary=True dans connection.cursor(dictionary=True) fait en sorte que les méthodes de récupération renvoient les lignes sous forme de dictionnaires (nom de colonne -> valeur) au lieu de tuples, ce qui est souvent plus pratique.

Opération UPDATE

Utilisez une instruction UPDATE avec execute(). N’oubliez pas d’utiliser des requêtes paramétrées et de valider la transaction.

# (En continuant, en supposant que 'conn' est ouverte)

def update_employee_salary(connection, employee_id, new_salary):
    """Met à jour le salaire pour un ID d'employé donné."""
    sql = "UPDATE employees SET salary = %s WHERE id = %s;"
    cursor = connection.cursor()
    try:
        cursor.execute(sql, (new_salary, employee_id))
        connection.commit()
        if cursor.rowcount > 0:
             print(f"Salaire mis à jour pour l'employé ID {employee_id}.")
        else:
             print(f"Aucun employé trouvé avec l'ID {employee_id}.")
    except Error as e:
        print(f"Échec de la mise à jour de l'enregistrement : {e}")
        connection.rollback()
    finally:
        if cursor:
            cursor.close()

# -- À l'intérieur du bloc d'exécution principal --
if conn and conn.is_connected():
    # ... (création, insertion, récupération) ...
    update_employee_salary(conn, 1, 60000.00) # Met à jour le salaire de l'employé avec l'ID 1
    # ... (fermeture de la connexion) ...

Opération DELETE

Utilisez une instruction DELETE avec execute(). Utilisez des requêtes paramétrées et validez.

# (En continuant, en supposant que 'conn' est ouverte)

def delete_employee(connection, employee_id):
    """Supprime un employé avec l'ID donné."""
    sql = "DELETE FROM employees WHERE id = %s;"
    cursor = connection.cursor()
    try:
        cursor.execute(sql, (employee_id,))
        connection.commit()
        if cursor.rowcount > 0:
             print(f"Employé avec l'ID {employee_id} supprimé.")
        else:
             print(f"Aucun employé trouvé avec l'ID {employee_id}.")
    except Error as e:
        print(f"Échec de la suppression de l'enregistrement : {e}")
        connection.rollback()
    finally:
        if cursor:
            cursor.close()

# -- À l'intérieur du bloc d'exécution principal --
if conn and conn.is_connected():
    # ... (création, insertion, récupération, mise à jour) ...
    delete_employee(conn, 3) # Supprime l'employé avec l'ID 3
    # ... (fermeture de la connexion) ...

Transactions (commit() et rollback())

Les transactions de base de données garantissent l’intégrité des données en utilisant les propriétés ACID (Atomicité, Cohérence, Isolation, Durabilité). Les opérations comme INSERT, UPDATE, DELETE font généralement partie d’une transaction.

Il est crucial de valider les modifications après des opérations DML réussies ou d’annuler en cas d’erreurs, comme montré dans les exemples INSERT, UPDATE et DELETE. Par défaut, mysql-connector-python pourrait avoir l’autocommit désactivé, nécessitant des validations explicites.

Déconnexion de la base de données (close())

Fermez toujours la connexion à la base de données lorsque vous avez terminé pour libérer les ressources.

if conn and conn.is_connected():
    conn.close()
    print("La connexion MySQL est fermée")

L’utilisation d’une instruction with pour la connexion et le curseur gère la fermeture automatiquement, même en cas d’erreurs.

# Modèle recommandé utilisant 'with'
try:
    with create_connection("localhost", "testuser", "test1234", "testdb") as conn:
        if conn and conn.is_connected():
            with conn.cursor(dictionary=True) as cursor:
                cursor.execute("SELECT * FROM employees WHERE age > %s", (30,))
                results = cursor.fetchall()
                for row in results:
                    print(row)
            # Le curseur est automatiquement fermé ici
        # La connexion est automatiquement fermée ici (ou annulée en cas d'erreur dans 'with')
        # Remarque : un commit pourrait encore être nécessaire explicitement selon les paramètres d'autocommit
except Error as e:
    print(f"Erreur de base de données : {e}")
except Exception as e:
    print(f"Une erreur s'est produite : {e}")

Gestion des erreurs

Les opérations de base de données peuvent échouer pour diverses raisons (SQL invalide, problèmes de connexion, violations de contraintes). Utilisez des blocs try...except pour intercepter les erreurs. Le module mysql-connector lève des exceptions dérivées de mysql.connector.Error.

Interceptez les erreurs spécifiques ou la classe de base Error selon les besoins.