Data Mapper Pattern in Enterprise Application Architecture

Introduction In the book "Patterns of Enterprise Application Architecture", Martin Fowler introduces the Data Mapper pattern, a crucial technique for applications that require a clean separation between business logic and data persistence. While other patterns (like Active Record) mix business logic and database operations into a single object, Data Mapper enforces a clear separation. What is the Data Mapper Pattern? The Data Mapper pattern: Maps domain objects to database structures and vice versa. Keeps business objects unaware of how their data is stored or retrieved. Enables better scaling for large enterprise applications. Benefits: High separation of concerns. Easier unit testing of business logic. Simplifies switching database technologies (e.g., from SQLite to PostgreSQL). Real-world Example in Python Without DataMapper import sqlite3 class Product: def init(self, id, name, price): self.id = id self.name = name self.price = price self.connection = sqlite3.connect("database.db") def save(self): cursor = self.connection.cursor() cursor.execute('''INSERT INTO products (id, name, price) VALUES (?, ?, ?)''', (self.id, self.name, self.price)) self.connection.commit() @classmethod def find(cls, product_id): connection = sqlite3.connect("database.db") cursor = connection.cursor() cursor.execute('''SELECT * FROM products WHERE id = ?''', (product_id,)) row = cursor.fetchone() if row: return Product(row[0], row[1], row[2]) return None @classmethod def list_all(cls): connection = sqlite3.connect("database.db") cursor = connection.cursor() cursor.execute('''SELECT * FROM products''') rows = cursor.fetchall() return [Product(row[0], row[1], row[2]) for row in rows] Imagine we want to manage a collection of products for a store. We'll use: A domain model Product. A mapper class ProductMapper. A simple SQLite database for persistence. 1. Domain Model (domain/product.py) class Product: def init(self, product_id: int, name: str, price: float): self.product_id = product_id self.name = name self.price = price def repr(self): return f"Product(id={self.product_id}, name='{self.name}', price={self.price})" 2. Data Mapper (mappers/product_mapper.py) import sqlite3 from app.domain.product import Product class ProductMapper: def init(self, connection: sqlite3.Connection): self.connection = connection self._create_table() def _create_table(self): cursor = self.connection.cursor() cursor.execute(''' CREATE TABLE IF NOT EXISTS products ( product_id INTEGER PRIMARY KEY, name TEXT NOT NULL, price REAL NOT NULL ) ''') self.connection.commit() def insert(self, product: Product): cursor = self.connection.cursor() cursor.execute(''' INSERT INTO products (product_id, name, price) VALUES (?, ?, ?) ''', (product.product_id, product.name, product.price)) self.connection.commit() def find(self, product_id: int) -> Product: cursor = self.connection.cursor() cursor.execute('SELECT product_id, name, price FROM products WHERE product_id = ?', (product_id,)) row = cursor.fetchone() if row: return Product(row) return None def list_all(self): cursor = self.connection.cursor() cursor.execute('SELECT product_id, name, price FROM products') rows = cursor.fetchall() return [Product(row) for row in rows] 3.Using the Mapper (main.py) import sqlite3 from app.domain.product import Product from app.mappers.product_mapper import ProductMapper def main(): connection = sqlite3.connect(":memory:") # In-memory database for testing mapper = ProductMapper(connection) # Inserting products mapper.insert(Product(1, "Laptop", 999.99)) mapper.insert(Product(2, "Smartphone", 499.50)) # Listing products products = mapper.list_all() for product in products: print(product) # Finding a product product = mapper.find(1) print(f"Found product: {product}") if name == "main": main() 4. Testing with Python (tests/test_product_mapper.py) import unittest import sqlite3 from app.domain.product import Product from app.mappers.product_mapper import ProductMapper class TestProductMapper(unittest.TestCase): def setUp(self): self.connection = sqlite3.connect(":memory:") self.mapper = ProductMapper(self.connection) def test_insert_and_find(self): product = Product(1, "Tablet",

Apr 27, 2025 - 21:28

Data Mapper Pattern in Enterprise Application Architecture

Introduction

In the book "Patterns of Enterprise Application Architecture", Martin Fowler introduces the Data Mapper pattern, a crucial technique for applications that require a clean separation between business logic and data persistence.

While other patterns (like Active Record) mix business logic and database operations into a single object, Data Mapper enforces a clear separation.

What is the Data Mapper Pattern?

The Data Mapper pattern:

Maps domain objects to database structures and vice versa.
Keeps business objects unaware of how their data is stored or retrieved.
Enables better scaling for large enterprise applications.

Benefits:

High separation of concerns.
Easier unit testing of business logic.
Simplifies switching database technologies (e.g., from SQLite to PostgreSQL).

Real-world Example in Python

Without DataMapper

import sqlite3

class Product:
    def __init__(self, id, name, price):
        self.id = id
        self.name = name
        self.price = price
        self.connection = sqlite3.connect("database.db") 
    def save(self):

        cursor = self.connection.cursor()
        cursor.execute('''INSERT INTO products (id, name, price) VALUES (?, ?, ?)''',
                       (self.id, self.name, self.price))
        self.connection.commit()

    @classmethod
    def find(cls, product_id):

        connection = sqlite3.connect("database.db")  
        cursor = connection.cursor()
        cursor.execute('''SELECT * FROM products WHERE id = ?''', (product_id,))
        row = cursor.fetchone()
        if row:
            return Product(row[0], row[1], row[2])
        return None

    @classmethod
    def list_all(cls):

        connection = sqlite3.connect("database.db")
        cursor = connection.cursor()
        cursor.execute('''SELECT * FROM products''')
        rows = cursor.fetchall()
        return [Product(row[0], row[1], row[2]) for row in rows]

Imagine we want to manage a collection of products for a store. We'll use:

A domain model Product.
A mapper class ProductMapper.
A simple SQLite database for persistence.

1. Domain Model (`domain/product.py`)

class Product:
    def __init__(self, product_id: int, name: str, price: float):
        self.product_id = product_id
        self.name = name
        self.price = price

    def __repr__(self):
        return f"Product(id={self.product_id}, name='{self.name}', price={self.price})"

2. Data Mapper (`mappers/product_mapper.py`)

import sqlite3
from app.domain.product import Product

class ProductMapper:
    def __init__(self, connection: sqlite3.Connection):
        self.connection = connection
        self._create_table()

    def _create_table(self):
        cursor = self.connection.cursor()
        cursor.execute('''
            CREATE TABLE IF NOT EXISTS products (
                product_id INTEGER PRIMARY KEY,
                name TEXT NOT NULL,
                price REAL NOT NULL
            )
        ''')
        self.connection.commit()

    def insert(self, product: Product):
        cursor = self.connection.cursor()
        cursor.execute('''
            INSERT INTO products (product_id, name, price) VALUES (?, ?, ?)
        ''', (product.product_id, product.name, product.price))
        self.connection.commit()

    def find(self, product_id: int) -> Product:
        cursor = self.connection.cursor()
        cursor.execute('SELECT product_id, name, price FROM products WHERE product_id = ?', (product_id,))
        row = cursor.fetchone()
        if row:
            return Product(*row)
        return None

    def list_all(self):
        cursor = self.connection.cursor()
        cursor.execute('SELECT product_id, name, price FROM products')
        rows = cursor.fetchall()
        return [Product(*row) for row in rows]

3.Using the Mapper (`main.py`)

import sqlite3
from app.domain.product import Product
from app.mappers.product_mapper import ProductMapper

def main():
    connection = sqlite3.connect(":memory:")  # In-memory database for testing
    mapper = ProductMapper(connection)

    # Inserting products
    mapper.insert(Product(1, "Laptop", 999.99))
    mapper.insert(Product(2, "Smartphone", 499.50))

    # Listing products
    products = mapper.list_all()
    for product in products:
        print(product)

    # Finding a product
    product = mapper.find(1)
    print(f"Found product: {product}")

if __name__ == "__main__":
    main()

4. Testing with Python (`tests/test_product_mapper.py`)

import unittest
import sqlite3
from app.domain.product import Product
from app.mappers.product_mapper import ProductMapper

class TestProductMapper(unittest.TestCase):
    def setUp(self):
        self.connection = sqlite3.connect(":memory:")
        self.mapper = ProductMapper(self.connection)

    def test_insert_and_find(self):
        product = Product(1, "Tablet", 299.99)
        self.mapper.insert(product)
        found = self.mapper.find(1)
        self.assertIsNotNone(found)
        self.assertEqual(found.name, "Tablet")

    def test_list_all(self):
        self.mapper.insert(Product(1, "Tablet", 299.99))
        self.mapper.insert(Product(2, "Monitor", 199.99))
        products = self.mapper.list_all()
        self.assertEqual(len(products), 2)

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

Conclusion

The Data Mapper pattern is an excellent choice for applications that prioritize business logic independence from database concerns.
By separating responsibilities, it enables easier evolution of the system, integration of caching layers, database migrations, and safe refactoring without impacting business logic.

Adopting patterns like Data Mapper will help you build cleaner, more robust, and scalable enterprise software.

link repository: https://github.com/Marant7/datamapperpython.git