Sauvegarder ses objets en base de données

En programmation orientée objet, une des pratiques courantes consiste à conserver nos données pour utilisation future. C’est ce qu’on appelle la persistance des données. Ces données peuvent être enregistrées sous différentes formes : fichiers JSON, utilisation du module pickle de Python, ou encore dans une base de données relationnelle, comme MySQL. Ce dernier est un système de gestion de base de données relationnelle très populaire, appréciée pour sa facilité d’intégration avec différents langages de programmation, dont Python. Dans cette section, nous allons explorer la manière de conserver nos objets dans une base de données MySQL.

Néanmoins, nous nous heurtons à un défi : comment mapper des objets Python, qui ne sont pas relationnels (et qui ne possède pas de clé primaire), à une base de données relationnelle ? La solution à ce problème réside généralement dans l’utilisation d’une architecture appelée Data Access Layer (DAL), qui fait office de pont entre nos objets et la base de données. 

Un DAL est un élément de notre application responsable de la communication avec la base de données. Lorsque nous utilisons une architecture DAL, nous ne communiquons pas directement avec la base de données. Au lieu de cela, nous travaillons avec une représentation de la base de données, qui se charge d’exécuter les opérations nécessaires. Une particularité remarquable de notre DAL est sa faculté à associer nos objets à la clé primaire de la base de données de manière transparente. Cette correspondance peut être gérée à l’aide d’un simple dictionnaire où la clé est un objet et la valeur est la clé primaire correspondante dans la base de données.

Dans le cadre de ce cours, nous utiliserons une classe abstraite qui permettra d’établir et de maintenir une connexion à la base de données. Cette classe utilise le design pattern « singleton » pour garantir qu’une seule instance de la connexion à la base de données soit créée et utilisée tout au long de l’application. De plus, nous utiliserons le patron de conception « repository » pour gérer l’accès aux données. Cela nous permettra de structurer notre code d’une manière organisée et cohérente.

Ce concept de « repository » prend en charge la responsabilité de gérer les opérations de la base de données. Il permet de maintenir la séparation des préoccupations en isolant la logique liée à l’accès aux données de la logique métier de l’application. Le rôle du « repository » est d’exposer des méthodes pour effectuer des opérations de base de données, comme la création, la lecture, la mise à jour et la suppression. Cependant, l’utilisation d’un « repository » impose une spécialisation des requêtes SQL et une attention particulière à la manière dont les objets sont reconstruits à partir des données récupérées.

L’interface de base

La classe abstraite Database que nous voyons ci-dessous est une réalisation de l’architecture DAL mentionnée précédemment. Elle implémente la logique de connexion à une base de données MySQL, et fournit une interface pour les opérations CRUD (création, lecture, mise à jour, suppression) sur les objets. Passons en revue les éléments clés de cette classe.

import mysql.connector
from abc import ABC, abstractmethod

class Database(ABC):
  _conn = None
  _registry = {}

  def __init__(self):
    if Database._conn is None:
      Database._conn = mysql.connector.connect(host="database.zaretti.be",
                                               user="YOUR_USERNAME_HERE",
                                               password="YOUR_PASSWORD_HERE",
                                               database="YOUR_DATABASE_HERE",
                                               port=33006)

  class Connect:
    def __init__(self):
      self.cursor = None

    def __enter__(self):
      self.cursor = Database._conn.cursor()
      return self.cursor

    def __exit__(self, type, lvalue, traceback):
      if type is None:
        Database._conn.commit()
      self.cursor.close()

  @abstractmethod
  def get(self, **kwargs):
    pass

  @abstractmethod
  def add(self, obj):
    pass

  @abstractmethod
  def update(self, obj):
    pass

  @abstractmethod
  def remove(self, obj):
    pass

Cette classe abstraite fournit point de départ pour créer une couche d’accès aux données dans une application Python. Elle établit une connexion unique à la base de données et définit une interface pour effectuer des opérations CRUD sur la base de données. Les classes héritant de cette classe abstraite peuvent ensuite fournir des implémentations spécifiques de ces opérations en fonction des besoins de l’application. Notez les quelques choix techniques suivant:

• Deux variables de classe _conn et _registry sont déclarées. _conn est utilisée pour maintenir la connexion à la base de données, et _registry est un dictionnaire qui servira à faire le lien entre nos objets et la clé primaire dans la base de données.

• Dans le constructeur de la classe, nous vérifions si une connexion à la base de données a déjà été établie. Si ce n’est pas le cas, nous créons une nouvelle connexion.

• La classe interne Connect est déclarée pour gérer la connexion à la base de données. Cette classe utilise le protocole de gestion de contexte de Python, qui permet d’utiliser la déclaration with pour gérer automatiquement l’ouverture et la fermeture du curseur et la gestion d’une transaction.

• Les méthodes get, add,  update et  remove sont déclarées comme des méthodes abstraites. Ces méthodes seront implémentées par chaque sous-classe qui hérite de Database pour spécifier comment les objets doivent être récupérés, ajoutés, mis à jour et supprimés de la base de données.

Dans le cadre de ce cours, l’objectif principal est d’apprendre à utiliser efficacement cette interface et de comprendre comment elle facilite la gestion de la persistance des objets dans une base de données. La mise en place d’une telle classe peut s’avérer complexe et demande une compréhension approfondie de certains concepts avancés de programmation orientée objet. Cependant, une fois que cette classe de base est mise en place, l’utiliser dans le reste de votre code est beaucoup plus simple et plus direct.

Ainsi, je vous recommande de copier-coller cette classe abstraite dans votre code. Ensuite, vous pouvez créer des sous-classes qui héritent de Database, et implémenter les méthodes get, add, update et remove pour chaque type d’objet spécifique que vous souhaitez gérer dans votre base de données. Cela vous permettra de vous concentrer sur l’apprentissage de la manière dont la persistance des objets est gérée en Python, sans vous préoccuper des détails de la gestion de la connexion à la base de données.

Implémentation de l’interface

Une fois que nous avons une classe abstraite comme notre Database, la prochaine étape est de l’implémenter pour un type d’objet spécifique. En d’autres termes, nous devons créer une classe qui hérite de Database et qui définit comment les méthodes abstraites fonctionnent pour ce type d’objet. C’est ce que nous appelons la spécialisation de l’interface. Pour illustrer cela, considérons l’exemple de la classe User ci-dessous :

Base de données

Code correspondant

Sql

CREATE TABLE `users` (
  `id` int NOT NULL,
  `first_name` varchar(32) NOT NULL,
  `last_name` varchar(32) NOT NULL,
  PRIMARY KEY (`id`)
);

UML

Code correspondant

Python

class User:
  def __init__(self, first_name, last_name):
    self.first_name = first_name
    self.last_name = last_name

La première étape pour créer notre implémentation spécifique de Database est de déclarer une nouvelle classe qui hérite de Database. Par convention, nous pourrions l’appeler UserDatabase. Dans UserDatabase, nous devons définir comment chaque méthode abstraite fonctionnera pour la classe User. Cela implique d’écrire des requêtes SQL spécifiques qui interagissent avec la table users de notre base de données et de spécifier comment les données retournées par ces requêtes sont converties en objets User.

Commençons par la méthode add. Nous pouvons écrire une requête SQL qui insère une nouvelle ligne dans la table users, en utilisant les attributs first_name et last_name de l’objet User passé à la méthode. Pour la méthode get, nous pourrions écrire une requête qui sélectionne une ligne de la table users sur la base de l’id passé en argument, et nous retournerions un nouvel objet User construit à partir des données retournées par cette requête.

class UserDatabase(Database):

  def add(self, obj):
    query = "INSERT INTO `users` (`id`, `first_name`, `last_name`) VALUES (NULL, %s, %s);"

    with Database.Connect() as connexion:
      connexion.execute(query, (obj.first_name, obj.last_name))

    Database._registry[obj] = connexion.lastrowid

    return True

La méthode commence par définir une requête SQL pour insérer une nouvelle ligne dans la table users. La requête contient deux marqueurs de paramètre (%s), qui seront remplacés par les valeurs des attributs first_name et last_name de l’objet User. Ensuite, la méthode ouvre une connexion à la base de données grâce au gestionnaire de contexte Connect défini dans la classe Database. Cela garantit que le curseur de la base de données est correctement fermé, même si une erreur se produit pendant l’exécution de la requête. La requête SQL est ensuite exécutée avec les valeurs des attributs first_name et last_name de l’objet User.

Après avoir exécuté la requête, la méthode ajoute une entrée au dictionnaire _registry de la classe Database. La clé de cette entrée est l’objet User lui-même, et la valeur est l’ID de la nouvelle ligne insérée dans la table users (qui est automatiquement générée par MySQL). Cela crée le lien entre l’objet User et son identifiant en base de données.

La méthode update fonctionne de manière similaire, mais elle exécutera une requête SQL UPDATE pour modifier les valeurs de la ligne correspondante dans la table users. La méthode remove, quant à elle, exécute une requête SQL DELETE pour supprimer la ligne correspondante de la table users. Après avoir supprimé la ligne de la base de données, la méthode remove doit également supprimer l’entrée correspondante du dictionnaire _registry.

class UserDatabase(Database):

  def remove(self, obj):
    query = "DELETE FROM users WHERE id = %s"

    with Database.Connect() as connexion:
      connexion.execute(query, (Database._registry[obj], ))

    Database._registry.pop(obj)
    return True

  def update(self, obj):
    query = "UPDATE users SET first_name=%s, last_name=%s WHERE id=%s"

    with Database.Connect() as connexion:
      connexion.execute(query, (obj.first_name, obj.last_name, Database._registry[obj], ))

Enfin, la méthode get qui est une méthode clé de notre classe UserDatabase. Cette fonction est utilisée pour récupérer un utilisateur de la base de données en fonction d’un critère spécifique, dans notre exemple un “id”, qui correspond à la clé primaire de la table des utilisateurs dans notre base de données. La clé primaire est unique pour chaque utilisateur, nous permettant d’identifier un utilisateur spécifique de manière fiable.

La méthode get est définie avec un argument **kwargs. C’est une syntaxe spéciale en Python qui permet de passer une variable en argument sous d’une fonction sous forme de paires clé-valeur. Dans notre cas, nous l’utilisons pour passer des critères de recherche à la méthode get. Nous pouvons donc l’utiliser comme ceci: user_database.get(id=1). C’est à l’intérieur de la méthode que nous vérifions si la clé “id” est présente dans kwargs. Si c’est le cas, nous construisons une requête SQL qui sélectionne l’utilisateur avec cet id dans la table des utilisateurs.

class UserDatabase(Database):

  def get(self, **kwargs):
    if "id" in kwargs:
      query = "SELECT id, first_name, last_name FROM users WHERE id = %s"

      with Database.Connect() as connexion:
        connexion.execute(query, (kwargs["id"], ))
        record = connexion.fetchone()

        if record is not None:
          obj = User(record[1], record[2])
          Database._registry[obj] = record[0]
          return obj

    return None

Comme pour les autres méthodes, nous ouvrons une connexion à la base de données. Ensuite, on récupère le premier enregistrement retourné par la requête avec fetchone(). Comme nous cherchons un utilisateur par son id, qui est une clé primaire unique dans la base de données, la requête ne retournera au maximum qu’un seul enregistrement. Si un enregistrement est retourné, un nouvel objet User est créé à partir des données de l’enregistrement. Cet objet est ensuite enregistré dans le registre de la base de données pour garder une trace de l’association entre l’objet et l’enregistrement de la base de données. Enfin, l’objet User est retourné. Si aucun enregistrement n’est retourné par la requête, la méthode retourne None.

Cela dit, cette méthode peut facilement être étendue pour chercher des utilisateurs sur la base d’autres critères, comme par exemple en appelant l’objet comme ceci: user_database.get(last_name="Zaretti"). Dans ce cas, comme plusieurs utilisateurs peuvent avoir le même nom de famille, la requête peut retourner plusieurs enregistrements et vous devriez utiliser fetchall() au lieu de fetchone(). Vous devrez également créer un objet User pour chaque enregistrement retourné et ajouter tous ces objets au registre.

class UserDatabase(Database):

  def get(self, **kwargs):
    if "id" in kwargs:
      pass
    elif "last_name" in kwargs:
      query = "SELECT id, first_name, last_name FROM users WHERE last_name = %s"
      objs = []

      with Database.Connect() as connexion:
        connexion.execute(query, (kwargs["last_name"],))
        for record in connexion.fetchall():
          obj = User(record[1], record[2])
          Database._registry[obj] = record[0]
          objs.append(obj)
      return objs
    return None

Cependant, n’oubliez pas que quel que soit le nombre d’objets que vous créez, il faut maintenir le registre à jour. Le registre est la clé pour garder un lien clair entre vos objets Python et les enregistrements correspondants dans la base de données. C’est lui qui vous permet de garder la synchronisation entre l’état de vos objets en mémoire et leur état persistant dans la base de données. 

De même, lorsque vous ajoutez ou supprimez un objet de la base de données, vous devez aussi l’ajouter ou le supprimer du registre. Laisser une référence à un objet supprimé dans le registre pourrait conduire à des erreurs et à une confusion, car vous auriez un objet en mémoire qui n’a pas de correspondance dans la base de données. Ainsi, chaque fois que vous effectuez une opération qui modifie la base de données, pensez toujours à mettre à jour le registre en conséquence. 

En conclusion, le registre maintient la cohérence entre le modèle d’objet Python et le schéma relationnel de la base de données. Les objets Python n’ont pas de clé primaire, contrairement aux enregistrements de la base de données. Le registre assure la liaison entre les deux, offrant un niveau d’abstraction qui permet aux objets de conserver leur indépendance tout en étant capables d’interagir efficacement avec la base de données.

Le code complet de ce chapitre, incluant la classe Database abstraite ainsi que l’exemple d’implémentation de l’interface avec la classe UserDatabase, est disponible dans le spoiler ci-dessous. Je vous encourage à explorer ce code en détail et d’essayer de l’adapter à vos besoins. Si vous souhaitez adapter ces concepts à vos propres classes, n’hésitez pas à copier-coller l’interface et à développer vos propres implémentations basées sur l’exemple donné ici. En réalité, les implémentations suivront très souvent la même structure, car elles sont basées sur le même modèle de conception : seuls l’objet concerné et la requête SQL changeront.

Par ailleurs, si vous souhaitez voir un exemple concret d’un projet utilisant cette architecture, n’hésitez pas à visiter ce lien. Vous y trouverez un exemple de projet mettant en pratique les concepts de ce chapitre.

import mysql.connector
from abc import ABC, abstractmethod


class Database(ABC):
  _conn = None
  _registry = {}

  def __init__(self):
    if Database._conn is None:
      Database._conn = mysql.connector.connect(host="database.zaretti.be",
                                               user="YOUR_USERNAME_HERE",
                                               password="YOUR_PASSWORD_HERE",
                                               database="YOUR_DATABASE_HERE",
                                               port=33006)

  class Connect:

    def __init__(self):
      self.cursor = None

    def __enter__(self):
      self.cursor = Database._conn.cursor()
      return self.cursor

    def __exit__(self, type, lvalue, traceback):
      if type is None:
        Database._conn.commit()
      self.cursor.close()

  @abstractmethod
  def get(self, **kwargs):
    pass

  @abstractmethod
  def add(self, obj):
    pass

  @abstractmethod
  def update(self, obj):
    pass

  @abstractmethod
  def remove(self, obj):
    pass

class UserDatabase(Database):

  def get(self, **kwargs):

    if "id" in kwargs:
      query = "SELECT id, first_name, last_name FROM users WHERE id = %s"
      obj = None

      with Database.Connect() as connexion:
        connexion.execute(query, (kwargs["id"], ))
        record = connexion.fetchone()

        if record is not None:
          obj = User(record[1], record[2])
          Database._registry[obj] = record[0]
      return obj
    elif "last_name" in kwargs:
      query = "SELECT id, first_name, last_name FROM users WHERE last_name = %s"
      objs = []

      with Database.Connect() as connexion:
        connexion.execute(query, (kwargs["last_name"],))
        for record in connexion.fetchall():
          obj = User(record[1], record[2])
          Database._registry[obj] = record[0]
          objs.append(obj)
      return objs
    else:
      query = "SELECT id, first_name, last_name FROM users"
      objs = []

      with Database.Connect() as connexion:
        connexion.execute(query)
        for record in connexion.fetchall():
          obj = User(record[1], record[2])
          Database._registry[obj] = record[0]
          objs.append(obj)

      return objs

  def add(self, obj):
    query = "INSERT INTO `users` (`id`, `first_name`, `last_name`) VALUES (NULL, %s, %s);"

    with Database.Connect() as connexion:
      connexion.execute(query, (obj.first_name, obj.last_name))

    Database._registry[obj] = connexion.lastrowid

    return True

  def remove(self, obj):
    query = "DELETE FROM users WHERE id = %s"

    with Database.Connect() as connexion:
      connexion.execute(query, (Database._registry[obj], ))

    Database._registry.pop(obj)

    return True

  def update(self, obj):
    query = "UPDATE users SET first_name=%s, last_name=%s WHERE id=%s"

    with Database.Connect() as connexion:
      connexion.execute(query, (
        obj.first_name,
        obj.last_name,
        Database._registry[obj],
      ))