Fonctionnalités de base

Avant de commencer, il convient d’importer le module, comme il est coutume de faire avec Python :

import sqlite3

Connexion

Cela fait, nous pouvons nous connecter à une BDD en utilisant la méthode connect et en lui passant l’emplacement du fichier de stockage en paramètre. Si ce dernier n’existe pas, il est alors créé :

connexion = sqlite3.connect("basededonnees.db") # BDD dans le fichier "basededonnees.db"

Comme vous pouvez le voir, nous récupérons un objet retourné par la fonction. Celui-ci est de type Connection et il nous permettra de travailler sur la base.

Par ailleurs, il est aussi possible de stocker la BDD directement dans la RAM en utilisant la chaîne clef ":memory:". Dans ce cas, il n’y aura donc pas de persistance des données après la déconnexion.

connexion = sqlite3.connect(":memory:") # BDD dans la RAM

C’est une bonne question ! Eh bien, premièrement ce qui est stocké dans la RAM est plus rapide d’accès que ce qu’il y a sur le disque dur. Ainsi, certains utiliseront la RAM de sorte à gagner en performance. Ensuite, les bases temporaires sont aussi très utiles pour effectuer des tests, par exemple des tests unitaires qui sont ainsi reproductibles aisément et n’altèrent pas d’éventuelles BDD persistantes.

Déconnexion

Que nous soyons connectés avec la RAM ou non, il ne faut pas oublier de nous déconnecter. Pour cela, il nous suffit de faire appel à la méthode close de notre objet Connection.

connexion.close() # Déconnexion

Un mot sur les types de champ

Comme nous allons bientôt voir comment exécuter des requêtes, il est important de connaître les types disponibles, avec leur correspondance en Python. Voici ci-dessous, un tableau récapitulatif :

Dans le sens inverse, les types Python du tableau seront utilisables avec leur correspondance SQLite. Il est vrai que la liste peut s’avérer restreignante. Heureusement, il est possible d’ajouter nos propres types de données.

Pour exécuter nos requêtes, nous allons nous servir d’un objet Cursor, récupéré en faisant appel à la méthode cursor de notre objet de type Connection.

curseur = connexion.cursor() # Récupération d'un curseur

Valider ou annuler les modifications

Lorsque nous effectuons des modifications sur une table (insertion, modification ou encore suppression d’éléments), celles-ci ne sont pas automatiquement validées. Ainsi, sans validation, les modifications ne sont pas effectuées dans la base et ne sont donc pas visibles par les autres connexions. Pour résoudre cela, il nous faut donc utiliser la méthode commit de notre objet de type Connection.

En outre, si nous effectuons des modifications puis nous souhaitons finalement revenir à l’état du dernier commit, il suffit de faire appel à la méthode rollback, toujours de notre objet de type Connection.

Voici un petit morceau de code résumant cela :

# modifications....
connexion.commit() # Validation des modifications
# modifications....
connexion.rollback() # Retour à l'état du dernier commit, les modifications effectuées depuis sont perdues

Exécuter une requête

Pour exécuter une requête il suffit de passer celle-ci à la méthode execute :

# Exécution unique
curseur.execute("""CREATE TABLE IF NOT EXISTS scores(
    id INTEGER PRIMARY KEY AUTOINCREMENT UNIQUE,
    pseudo TEXT,
    valeur INTEGER
)""")

Comme vous pouvez le voir, nous venons d’ajouter une table scores dans notre base jusqu’à présent vide.

Exécuter plusieurs requêtes

Pour exécuter plusieurs requêtes, comme pour ajouter des éléments à une table par exemple, nous pouvons faire appel plusieurs fois à la méthode execute :

donnees = [("toto", 1000), ("tata", 750), ("titi", 500)]
# Exécutions multiples
for donnee in donnees:
    curseur.execute("INSERT INTO scores (pseudo, valeur) VALUES (?, ?)", donnee)
connexion.commit() # Ne pas oublier de valider les modifications

Ou nous pouvons aussi passer par la méthode executemany :

donnees = [("toto", 1000), ("tata", 750), ("titi", 500)]
# Exécutions multiples
curseur.executemany("INSERT INTO scores (pseudo, valeur) VALUES (?, ?)", donnees)
connexion.commit() # Ne pas oublier de valider les modifications

Remarquez que nous utilisons ici, l’opérateur ? couplé à des tuples pour passer des paramètres aux requêtes, mais nous pouvons aussi utiliser des dictionnaires et l’opérateur : avec le nom des clefs :

donnees = (
    {"psd": "toto", "val": 1000},
    {"psd": "tata", "val": 750},
    {"psd": "titi", "val": 500}
)
# Exécutions multiples
curseur.executemany("INSERT INTO scores (pseudo, valeur) VALUES (:psd, :val)", donnees)
connexion.commit() # Ne pas oublier de valider les modifications

Exécuter un script

Enfin, il est aussi possible d’exécuter un script directement à l’aide de la méthode executescript. Si celui-ci contient plusieurs requêtes, celles-ci doivent être séparées par des points-virgules.

# Exécution d'un script
curseur.executescript("""
    DROP TABLE IF EXISTS scores;

    CREATE TABLE scores(
    id INTEGER PRIMARY KEY AUTOINCREMENT UNIQUE,
    pseudo TEXT,
    valeur INTEGER);
    
    INSERT INTO scores(pseudo, valeur) VALUES ("toto", 1000);
    INSERT INTO scores(pseudo, valeur) VALUES ("tata", 750);
    INSERT INTO scores(pseudo, valeur) VALUES ("titi", 500)
""")
connexion.commit() # Ne pas oublier de valider les modifications

Pour récupérer des éléments, nous allons évidemment passer par une requête SQL. Il faudra ensuite parcourir le résultat et nous nous servirons de notre objet de type Cursor pour cela.

Mais avant de le faire, reprenons notre table scores et ajoutons quelques éléments afin d’avoir un exemple pratique.

Le code :

import sqlite3

# Connexion
connexion = sqlite3.connect('basededonnees.db')

# Récupération d'un curseur
curseur = connexion.cursor()

# Création de la table scores
curseur.executescript("""
    DROP TABLE IF EXISTS scores;

    CREATE TABLE scores(
    id INTEGER PRIMARY KEY AUTOINCREMENT UNIQUE,
    pseudo TEXT,
    valeur INTEGER);
""")

# Suppression des éléments de scores
curseur.execute("DELETE FROM scores")

# Préparation des données à ajouter
donnees = [
    ("toto", 1000),
    ("tata", 750),
    ("titi", 500),
    ("toto", 250),
    ("tata", 150),
    ("tete", 0)
    ]

# Insertion des données
curseur.executemany("INSERT INTO scores (pseudo, valeur) VALUES (?, ?)", donnees)

# Validation
connexion.commit()

# Parcours des enregistrements....

# Déconnexion
connexion.close()

Le résultat :

Un résultat à la fois

Pour parcourir un résultat à la fois, il suffit d’utiliser la méthode fetchone qui retourne un résultat sous forme de tuple, ou None, s’il n’y en a pas.

donnee = ("titi", )
curseur.execute("SELECT valeur FROM scores WHERE pseudo = ?", donnee)
print(curseur.fetchone()) # Affiche "(500,)"

Plusieurs résultats d’un coup

Vous comprendrez que cette technique montre vite ses limites quand le nombre de résultats augmente, et ce même si nous pouvons procéder ainsi :

donnee = ("tata", )
curseur.execute("SELECT valeur FROM scores WHERE pseudo = ?", donnee)
result = curseur.fetchone()
while result:
    print(result)
    result = curseur.fetchone()
# Affiche "(750,)" puis "(150,)"

Or, fetchmany, utilisable de la même manière, permet justement de récupérer plusieurs résultats d’un coup. Le nombre de résultats prend par défaut la valeur de l’attribut arraysize du curseur, mais nous pouvons aussi passer un nombre à la méthode. S’il n’y a pas de résultat, la liste retournée est vide :

print(curseur.arraysize) # Affiche "1"
donnee = (400, )

curseur.execute("SELECT pseudo FROM scores WHERE valeur > ?", donnee)
print(curseur.fetchmany()) # Affiche "[('toto',)]"
print(curseur.fetchmany()) # Affiche "[('tata',)]"

curseur.execute("SELECT pseudo FROM scores WHERE valeur > ?", donnee)
print(curseur.fetchmany(2)) # Affiche "[('toto',), ('tata',)]"

Comme vous pouvez le constater, cela revient à utiliser fetchone si l’attribut arraysize du curseur vaut 1, ce qui n’est pas très utile.

Tout ou rien

Enfin, pour récupérer directement tous les résultats d’une requête, nous pouvons faire appel à la méthode fetchall. Là encore, elle retourne une liste vide s’il n’y a pas de résultats.

curseur.execute("SELECT * FROM scores")
resultats = curseur.fetchall()
for resultat in resultats:
    print(resultat)

Par ailleurs, nous pouvons aussi utiliser le curseur comme un itérable :

curseur.execute("SELECT * FROM scores")
for resultat in curseur:
    print(resultat)

Les deux codes ont le même effet et affichent :

(1, "toto", 1000)
(2, "tata", 750)
(3, "titi", 500)
(4, "toto", 250)
(5, "tata", 150)
(6, "tete", 0)

Avec sqlite3, nous pouvons récupérer quelques informations sur l’état actuel de notre base.

En transaction ou pas

Tout d’abord, pour savoir si des modifications ont été apportées sans être validées, il suffit de récupérer la valeur de l’attribut in_transaction de notre objet de type Connection. En effet, celui-ci vaut True si c’est le cas et False sinon.

# modifications...
print(connexion.in_transaction) # Affiche "True"
connexion.commit()
print(connexion.in_transaction) # Affiche "False"

Connaître le nombre de modifications depuis le dernier commit

Ensuite, pour être au courant du nombre de modifications (ajouts, mises à jour ou suppressions) apportées depuis notre connexion à la base, il suffit de récupérer la valeur de l’attribut total_changes de notre objet de type Connection.

Dans l’exemple ci-dessous, nous insérons autant de scores qu’il y a de lettres dans la chaîne de caractères :

print(connexion.total_changes) # Affiche "0"
chaine = "azertyuiopmlkjhgfdsqwxcvbnmlkjhgfdsqazertyuiopnbvcxw"
print(len(chaine)) # Affiche "52"
for donnee in enumerate(chaine):
    curseur.execute("INSERT INTO scores (pseudo, valeur) VALUES (?, ?)", donnee[::-1])
print(connexion.total_changes) # Affiche "52"

Connaitre le nombre de lignes impactées par une exécution

De même, pour connaître le nombre de lignes impactées par une exécution, il suffit d’utiliser l’attribut rowcount de notre objet de type Cursor. S’il n’y a eu aucune exécution ou que le nombre de de lignes ne peut pas être déterminé (comme pour une sélection par exemple), il vaut -1. De plus, pour les versions de SQLite antérieure à la 3.6.5, la valeur vaut 0 après une suppression totale des éléments d’une table.

Voici un exemple :

print(curseur.rowcount) # Affiche "-1"

donnee = ("toto", 1000)
curseur.execute("INSERT INTO scores (pseudo, valeur) VALUES (?, ?)", donnee)
print(curseur.rowcount) # Affiche "1"

donnees = [("tata", 750), ("titi", 500)]
curseur.executemany("INSERT INTO scores (pseudo, valeur) VALUES (?, ?)", donnees)
print(curseur.rowcount) # Affiche "2"

curseur.execute("SELECT * FROM scores")
print(curseur.rowcount) # Affiche "-1"

curseur.execute("DELETE FROM scores")
print(curseur.rowcount) # Affiche "9" (0 si version SQLite < 3.6.5)

Récupérer l’identifiant de la dernière ligne insérée

Par ailleurs, nous pouvons aussi récupérer l’identifiant du dernier enregistrement dans une table à l’aide de l’attribut lastrowid de notre objet de type Connection :

from random import randint

# ...

donnee = (randint(1, 1000), "toto", 1000)
print(donnee[0]) # Affiche "589"
curseur.execute("INSERT INTO scores (id, pseudo, valeur) VALUES (?, ?, ?)", donnee)
curseur.execute("SELECT * FROM scores WHERE id = ?", (curseur.lastrowid, ))
print(curseur.fetchone()) # Affiche "(589, 'toto', 1000)"

Dans l’exemple ci-dessus, nous insérons un enregistrement avec un identifiant aléatoire puis nous récupérons ce même enregistrement grâce à la valeur de l’attribut.

Dès que le nombre de tables augmente, il est souvent primordial de les lier à l’aide de clefs étrangères.

Activer les clefs étrangères

Avec sqlite3, les clefs étrangères ne sont pas activées de base. Il nous faut donc y remédier avec la requête adéquate :

curseur.execute("PRAGMA foreign_keys = ON") # Active les clés étrangères

Lier deux tables

Maintenant que c’est fait, nous pouvons ajouter une table joueurs, donc créer une nouvelle table scores (veillez à supprimer l’ancienne si jamais), puis remplir celles-ci et récupérer les enregistrements, avec une bonne utilisation des clefs étrangères :

Le code :

import sqlite3

# Connexion
connexion = sqlite3.connect("basededonnees.db") 

# Récupération d'un curseur
curseur = connexion.cursor()

# Activation clés étrangères
curseur.execute("PRAGMA foreign_keys = ON") 

# Création table joueur puis score si elles n'existent pas encore
# puis suppression des données dans joueurs (et dans scores aussi par cascade)
# afin d'éviter les répétitions d'enregistrements avec des exécutions multiples
curseur.executescript("""
    DROP TABLE IF EXISTS scores;
    DROP TABLE IF EXISTS joueurs;
    
    CREATE TABLE joueurs(
    id_joueur INTEGER PRIMARY KEY,
    pseudo TEXT,
    mdp TEXT);

    CREATE TABLE scores(
    id_score INTEGER PRIMARY KEY,
    fk_joueur INTEGER NOT NULL,
    valeur INTEGER,
    FOREIGN KEY(fk_joueur) REFERENCES joueurs(id_joueur)
    ON DELETE CASCADE);
""")

# Préparation des données
donnees_joueur = [
    ("toto", "123"),
    ("tata", "azerty"),
    ("titi", "qwerty")
]
donnees_score = [
    (1, 1000),
    (2, 750),
    (3, 500)
]

# Insertion des données dans table joueur puis score
curseur.executemany("INSERT INTO joueurs (pseudo, mdp) VALUES (?, ?)", donnees_joueur)
curseur.executemany("INSERT INTO scores (fk_joueur, valeur) VALUES (?, ?)", donnees_score)

# Validation des ajouts
connexion.commit()

# Affichage des données
for joueur in curseur.execute("SELECT * FROM joueurs"):
    print("joueur :", joueur)

for score in curseur.execute("SELECT * FROM scores"):
    print("score :", score)

# Déconnexion
connexion.close()

Le résultat :

joueurs

scores

joueur : (1, 'toto', '123')
joueur : (2, 'tata', 'azerty')
joueur : (3, 'titi', 'qwerty')
score : (1, 1, 1000)
score : (2, 2, 750)
score : (3, 3, 500)

Vous remarquerez que les mots de passe ne sont pas chiffrés ce qui, comme vous le savez, est une pratique fortement déconseillée.

Une fois, nos tables créées et remplies, nous pouvons facilement travailler dessus à l’aide de jointures, comme pour récupérer le meilleur score (pseudo et valeur) par exemple :

# Récupération du meilleur score
curseur.execute("""SELECT j.pseudo, s.valeur FROM joueurs as j INNER JOIN
    scores as s ON j.id_joueur = s.fk_joueur
    ORDER BY s.valeur DESC LIMIT 1""")
print(curseur.fetchone()) # Affiche "('toto', 1000)"

Dans l’exemple ci-dessus, nous utilisons le type de join le plus répandu (le INNER JOIN), mais nous aurions pu en utiliser d’autres.