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",

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] 

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})"

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]

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()

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()

Related Posts

How to Harden Your Node.js APIs – Security Best Practices

Apr 25, 2025  0

Digital Clock

Apr 20, 2025  0

Exploring a Jupyter Notebook Project on GitHub

Apr 21, 2025  0