Enterprise Design Patterns: Data Mapper with Python

Introduction

In the development of enterprise applications, choosing the right design patterns is crucial to creating scalable, maintainable, and testable systems. One such design pattern is the Data Mapper. This pattern separates the business logic of an application from the data persistence logic, which is especially useful when interacting with databases.

In this article, we will explore the Data Mapper pattern, describe its benefits, and implement it using Python to manage user data persistence in an SQLite database. We will also provide a real-world example to help you understand how to apply the Data Mapper pattern in your own applications.

What is the Data Mapper Pattern?

The Data Mapper pattern is a structural pattern that acts as a middle layer between an application's business logic (or domain objects) and the database. It provides the necessary abstraction to ensure that the domain objects are independent of the underlying database.

Key Characteristics

• Decoupling: Business objects are not dependent on the database schema or persistence mechanisms.

• Data Management: The Data Mapper is responsible for saving and retrieving objects from the database.

• Flexibility: By decoupling the data access logic, it becomes easier to swap or change database implementations without affecting the business logic.

Advantages of the Data Mapper Pattern

1. Persistence Independence: The application’s core logic (business objects) does not need to know how the data is stored.

2. Easy Unit Testing: Business logic can be tested without requiring access to the database, facilitating easier and faster unit testing.

3. Database Independence: The persistence layer can change without impacting the business objects, allowing more flexibility in choosing or changing databases.

4. Maintainability: With the separation of concerns, it is easier to maintain and scale the application.

Implementing the Data Mapper Pattern in Python

Step 1: Requiremensts (requiremensts.txt)

sqlite3

Step 2: Create the Database (/database/db_connection.py)

First, we need to create an SQLite database with a users table. SQLite is a lightweight relational database that doesn’t require additional setup, making it a great choice for demonstration purposes.

import sqlite3

def get_connection():
    connection = sqlite3.connect(":memory:")
    cursor = connection.cursor()

    cursor.execute('''
    CREATE TABLE users (
        id INTEGER PRIMARY KEY,
        name TEXT NOT NULL,
        email TEXT NOT NULL
    )
    ''')
    connection.commit()

    return connection

Step 3: Define the User Class (/models/user.py)

The User class represents a user entity, containing basic user data such as id, name and email. This class will not contain any database logic; it simply holds the data.

class User:
    def __init__(self, user_id, name, email):
        self.user_id = user_id
        self.name = name
        self.email = email

    def __str__(self):
        return f"User({self.user_id}, {self.name}, {self.email})"

Step 4: Create the UserMapper Class (/Mappers/ UserMapper.py)

The UserMapper class is where the Data Mapper pattern comes into play. This class will handle the operations of saving and retrieving user data from the database.

import sqlite3
from models.user import User

class UserMapper:
    def __init__(self, connection):
        self.connection = connection

    def insert(self, user):
        cursor = self.connection.cursor()
        cursor.execute(
            "INSERT INTO users (name, email) VALUES (?, ?)",
            (user.name, user.email)
        )
        self.connection.commit()

        user.user_id = cursor.lastrowid

    def find_by_id(self, user_id):
        cursor = self.connection.cursor()
        cursor.execute(
            "SELECT id, name, email FROM users WHERE id = ?",
            (user_id,)
        )
        row = cursor.fetchone()
        if row:
            return User(row[0], row[1], row[2])
        return None

Step 5: Handle User Creation and Display (app.py)

Next, we create functionality to allow adding users and viewing all users. Users will be added infinitely until the user chooses to stop, and then all users in the database will be displayed.

from database.db_connection import get_connection
from models.user import User
from mappers.user_mapper import UserMapper

connection = get_connection()

mapper = UserMapper(connection)

def add_users():
    while True:
        cursor = connection.cursor()
        cursor.execute("SELECT MAX(id) FROM users")
        max_id = cursor.fetchone()[0]
        next_id = max_id + 1 if max_id is not None else 1

        name = input("Introduce el nombre del usuario: ")
        email = input("Introduce el correo electrónico del usuario: ")

        new_user = User(next_id, name, email)

        mapper.insert(new_user)
        print(f"Usuario {new_user.name} agregado correctamente con ID {new_user.user_id}.")

        another = input("¿Quieres agregar otro usuario? (s/n): ")
        if another.lower() != 's':
            break

    show_all_users()

def show_all_users():
    print("\nUsuarios almacenados:")
    cursor = connection.cursor()
    cursor.execute("SELECT id, name, email FROM users")
    rows = cursor.fetchall()
    for row in rows:
        print(f"ID: {row[0]}, Nombre: {row[1]}, Correo: {row[2]}")

add_users()

Step 6: Unit Tests ( /tests/test_user_mapper.py)

Finally, we create unit tests to ensure that the UserMapper is functioning correctly. These tests check that users can be inserted and retrieved from the database.

import unittest
import sqlite3
from models.user import User
from mappers.user_mapper import UserMapper

class TestUserMapper(unittest.TestCase):

    @classmethod
    def setUpClass(cls):
        cls.connection = sqlite3.connect(':memory:')
        cursor = cls.connection.cursor()
        cursor.execute("CREATE TABLE users (id INTEGER PRIMARY KEY AUTOINCREMENT, name TEXT, email TEXT)")
        cls.connection.commit()
        cls.mapper = UserMapper(cls.connection)

    @classmethod
    def tearDownClass(cls):
        cls.connection.close()

    def test_insert_user(self):
        new_user = User(1, "John Doe", "john.doe@example.com")

        self.mapper.insert(new_user)

        cursor = self.connection.cursor()
        cursor.execute("SELECT id, name, email FROM users WHERE id=?", (new_user.user_id,))
        row = cursor.fetchone()
        self.assertIsNotNone(row)
        self.assertEqual(row[0], new_user.user_id)
        self.assertEqual(row[1], new_user.name)
        self.assertEqual(row[2], new_user.email)

    def test_find_by_id(self):
        new_user = User(1, "Jane Doe", "jane.doe@example.com")

        self.mapper.insert(new_user)

        retrieved_user = self.mapper.find_by_id(1)
        self.assertIsNotNone(retrieved_user)
        self.assertEqual(retrieved_user.user_id, new_user.user_id)
        self.assertEqual(retrieved_user.name, new_user.name)
        self.assertEqual(retrieved_user.email, new_user.email)

    def test_find_non_existent_user(self):
        retrieved_user = self.mapper.find_by_id(999)
        self.assertIsNone(retrieved_user)

if __name__ == '__main__':
    unittest.main()

Step 7: Run the app.py

Now, open a terminal or command prompt, navigate to the directory containing app.py, and run the following command:

python app.py

GitHub Link: https://github.com/andresebast161616/DataMapper.git

Conclusion

The Data Mapper pattern is an essential design pattern for building scalable and maintainable enterprise applications. By decoupling business logic from data persistence, the Data Mapper simplifies database operations and improves flexibility. It is especially beneficial for systems that require complex database interactions and frequent changes to the underlying database structure.

In this article, we demonstrated how to implement the Data Mapper pattern in Python using SQLite, focusing on managing user persistence. The unit tests provided ensure that the Data Mapper functions correctly, allowing users to be added to and retrieved from the database efficiently.

By applying this pattern, your application can achieve greater maintainability, testability, and scalability, making it an excellent choice for large enterprise systems.