Procédure d'utilisation

Pour interagir avec une base de données MySQL via Python, suivez cette séquence : établir une connnexion → créer un curseur → exécuter des requêtes...

# Installation du module : pip3 install pymysql
import pymysql

# Établissement de la connexion à la base de données
conn = pymysql.connect(
    host="127.0.0.1",
    port=3306,
    database="day35",
    user="root",
    password="123456"
)

# Création d'un curseur pour exécuter les requêtes
cursor = conn.cursor(cursor=pymysql.cursors.DictCursor)
# Utilisation de DictCursor permet d'obtenir les résultats sous forme de dictionnaires

rows = cursor.execute("SELECT * FROM ttt;")
print(rows)

result = cursor.fetchall()  # Récupération de tous les enregistrements
print(result)

cursor.close()
conn.close()

Insertion de plusieurs enregistrements

import pymysql

# Connexion à la base de données
conn = pymysql.connect(
    host="127.0.0.1",
    port=3306,
    database="day35",
    user="root",
    password="123456"
)
cursor = conn.cursor(cursor=pymysql.cursors.DictCursor)

# Suppression et création de la table 'sites'
cursor.execute('DROP TABLE IF EXISTS sites')
cursor.execute('CREATE TABLE sites(name VARCHAR(20) PRIMARY KEY, url VARCHAR(200))')

# Insertion multiple d'enregistrements
sql = "INSERT INTO sites(name, url) VALUES (%s, %s)"
data = [
    ("Google", "https://www.google.com"),
    ("Github", "https://www.github.com"),
    ("Taobao", "https://www.taobao.com")
]

cursor.executemany(sql, data)

# Validation de la transaction
conn.commit()
print(cursor.rowcount, "enregistrements insérés avec succès")

# Fermeture du curseur
cursor.close()

# Requête pour afficher les données insérées
cursor = conn.cursor()
cursor.execute('SELECT * FROM sites')
records = cursor.fetchall()
for record in records:
    print(record)

# Fermeture finale
cursor.close()
conn.close()

Encapsulation dans une fonction

Voici un exemple de fonction pour insérer des données dans une base de données :

import pymysql

def insert_data(entry_dict: dict):
    """
    Insère un enregistrement dans la base de données.
    
    Args:
        entry_dict (dict): Dictionnaire contenant les valeurs à insérer.
    """
    db = pymysql.connect(
        host="192.168.49.160",
        port=3306,
        database="day01",
        user="liqi",
        password="liqi123456"
    )
    cursor = db.cursor(cursor=pymysql.cursors.DictCursor)

    # Préparation de la requête SQL
    sql = "INSERT INTO t1(test_type, produce_name) VALUES (%s, %s)"
    
    # Extraction des valeurs depuis le dictionnaire
    test_type = entry_dict.get('test_type')
    produce_name = entry_dict.get('produce_name')

    # Données à insérer
    values = [test_type, produce_name]
    
    try:
        cursor.execute(sql, values)
        db.commit()
    except:
        db.rollback()

    print(cursor.rowcount, "enregistrements insérés")
    cursor.close()

if __name__ == '__main__':
    data = {
        'test_type': 'gemm',
        'produce_name': "Liqi1234"
    }
    insert_data(data)

Récupération du dernier enregistrement inséré

Voici comment récupérer le dernier enregistrement après une insertion :

import pymysql

def insert_and_fetch_last(entry_dict: dict):
    """
    Insère un enregistrement puis récupère le dernier ajouté.
    """
    db = pymysql.connect(
        host="192.168.49.160",
        port=3306,
        database="day01",
        user="liqi",
        password="liqi123456"
    )
    cursor = db.cursor(cursor=pymysql.cursors.DictCursor)

    # Requête d'insertion
    sql = "INSERT INTO t1(test_type, produce_name) VALUES (%s, %s)"
    test_type = entry_dict.get('test_type')
    produce_name = entry_dict.get('produce_name')
    
    values = [test_type, produce_name]
    
    try:
        cursor.execute(sql, values)
        db.commit()
    except:
        db.rollback()

    print(cursor.rowcount, "enregistrements insérés")

    # Récupération du dernier enregistrement inséré
    last_id = cursor.lastrowid - 1  # Ajustement pour l'index
    
    cursor = db.cursor(cursor=pymysql.cursors.DictCursor)
    select_query = "SELECT * FROM t1"
    cursor.execute(select_query)
    cursor.scroll(last_id, mode="absolute")
    
    latest_record = cursor.fetchone()
    print(latest_record)
    
    cursor.close()

if __name__ == '__main__':
    data = {
        'test_type': 'gemm',
        'produce_name': "Liqi1234"
    }
    insert_and_fetch_last(data)

Références : Ouvrage « Python Interview Guide » par Yuan Yuan Home – Chapitre 5 : Base de données.