cleveragents-core/.opencode/skills/programming-patterns/references/data-access/data-mapper.md

Data Mapper Pattern

Problem

In complex domains, coupling persistence logic directly to domain objects (as Active Record does) creates tight coupling between the domain model and database schema. Domain objects become harder to test, harder to change, and accumulate responsibilities they shouldn't have.

Solution

Introduce a separate mapper layer that transfers data between domain objects and the database. Domain objects have zero knowledge of the database. Mapper objects know how to load from and save to the database, performing the translation between the two representations.

When to Use

• Complex domain models with rich business logic
• When domain model and database schema differ significantly
• When you need to test domain logic without a database
• Systems following clean architecture or hexagonal architecture
• When the same domain model maps to multiple storage backends

When to Avoid

• Simple CRUD applications where Active Record suffices
• Prototypes where development speed is the priority
• When the domain model closely mirrors the database schema
• Small teams where the abstraction overhead isn't justified

Pseudocode

class User:                    # Pure domain object — no DB knowledge
    name, email

class UserMapper:              # Handles all persistence
    function find(id):
        row = db.query("SELECT * FROM users WHERE id = ?", id)
        return User(row.name, row.email)

    function save(user):
        if user.id:
            db.execute("UPDATE users SET ... WHERE id = ?", user)
        else:
            db.execute("INSERT INTO users ...", user)

    function delete(user):
        db.execute("DELETE FROM users WHERE id = ?", user.id)

Python Implementation

import sqlite3
from dataclasses import dataclass, field
from typing import Optional


# --- Domain Model (no DB knowledge) ---

@dataclass
class User:
    name: str
    email: str
    id: Optional[int] = field(default=None)

    def change_email(self, new_email: str):
        if "@" not in new_email:
            raise ValueError(f"Invalid email: {new_email}")
        self.email = new_email

    def __repr__(self):
        return f"User(id={self.id}, name={self.name!r}, email={self.email!r})"


# --- Data Mapper (handles persistence) ---

class UserMapper:
    def __init__(self, connection: sqlite3.Connection):
        self.conn = connection
        self.conn.row_factory = sqlite3.Row

    def create_table(self):
        self.conn.execute("""
            CREATE TABLE IF NOT EXISTS users (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                name TEXT NOT NULL,
                email TEXT NOT NULL
            )
        """)
        self.conn.commit()

    def save(self, user: User) -> User:
        if user.id is not None:
            self.conn.execute(
                "UPDATE users SET name = ?, email = ? WHERE id = ?",
                (user.name, user.email, user.id),
            )
        else:
            cursor = self.conn.execute(
                "INSERT INTO users (name, email) VALUES (?, ?)",
                (user.name, user.email),
            )
            user.id = cursor.lastrowid
        self.conn.commit()
        return user

    def find_by_id(self, user_id: int) -> Optional[User]:
        row = self.conn.execute(
            "SELECT * FROM users WHERE id = ?", (user_id,)
        ).fetchone()
        if row is None:
            return None
        return self._to_domain(row)

    def find_all(self) -> list[User]:
        rows = self.conn.execute("SELECT * FROM users").fetchall()
        return [self._to_domain(r) for r in rows]

    def delete(self, user: User) -> None:
        self.conn.execute("DELETE FROM users WHERE id = ?", (user.id,))
        self.conn.commit()
        user.id = None

    def _to_domain(self, row: sqlite3.Row) -> User:
        return User(id=row["id"], name=row["name"], email=row["email"])


# --- Demo ---
conn = sqlite3.connect(":memory:")
mapper = UserMapper(conn)
mapper.create_table()

# Create
alice = User(name="Alice", email="alice@example.com")
mapper.save(alice)
print(f"Saved: {alice}")

bob = User(name="Bob", email="bob@example.com")
mapper.save(bob)
print(f"Saved: {bob}")

# Read
found = mapper.find_by_id(1)
print(f"Found: {found}")

# Domain logic (no DB involved)
alice.change_email("alice@newdomain.com")
mapper.save(alice)
print(f"Updated: {alice}")

# List
print(f"All: {mapper.find_all()}")

# Delete
mapper.delete(bob)
print(f"After delete: {mapper.find_all()}")

conn.close()

Output:

Saved: User(id=1, name='Alice', email='alice@example.com')
Saved: User(id=2, name='Bob', email='bob@example.com')
Found: User(id=1, name='Alice', email='alice@example.com')
Updated: User(id=1, name='Alice', email='alice@newdomain.com')
All: [User(id=1, name='Alice', email='alice@newdomain.com'), User(id=2, name='Bob', email='bob@example.com')]
After delete: [User(id=1, name='Alice', email='alice@newdomain.com')]

Go Implementation

package main

import (
	"fmt"
	"sync"
)

// --- Domain Model (no DB knowledge) ---

type User struct {
	ID    int
	Name  string
	Email string
}

func (u User) String() string {
	return fmt.Sprintf("User(id=%d, name=%q, email=%q)", u.ID, u.Name, u.Email)
}

// --- Data Mapper ---

type UserMapper struct {
	mu     sync.Mutex
	store  map[int]*User
	nextID int
}

func NewUserMapper() *UserMapper {
	return &UserMapper{store: make(map[int]*User), nextID: 1}
}

func (m *UserMapper) Save(u *User) *User {
	m.mu.Lock()
	defer m.mu.Unlock()
	if u.ID == 0 {
		u.ID = m.nextID
		m.nextID++
	}
	// Store a copy to simulate DB round-trip
	copy := *u
	m.store[u.ID] = &copy
	return u
}

func (m *UserMapper) FindByID(id int) (*User, bool) {
	m.mu.Lock()
	defer m.mu.Unlock()
	u, ok := m.store[id]
	if !ok {
		return nil, false
	}
	copy := *u
	return &copy, true
}

func (m *UserMapper) FindAll() []*User {
	m.mu.Lock()
	defer m.mu.Unlock()
	result := make([]*User, 0, len(m.store))
	for _, u := range m.store {
		copy := *u
		result = append(result, &copy)
	}
	return result
}

func (m *UserMapper) Delete(u *User) {
	m.mu.Lock()
	defer m.mu.Unlock()
	delete(m.store, u.ID)
	u.ID = 0
}

func main() {
	mapper := NewUserMapper()

	alice := mapper.Save(&User{Name: "Alice", Email: "alice@example.com"})
	fmt.Println("Saved:", alice)

	bob := mapper.Save(&User{Name: "Bob", Email: "bob@example.com"})
	fmt.Println("Saved:", bob)

	found, _ := mapper.FindByID(1)
	fmt.Println("Found:", found)

	alice.Email = "alice@newdomain.com"
	mapper.Save(alice)
	fmt.Println("Updated:", alice)

	fmt.Print("All: [")
	for i, u := range mapper.FindAll() {
		if i > 0 {
			fmt.Print(", ")
		}
		fmt.Print(u)
	}
	fmt.Println("]")

	mapper.Delete(bob)
	fmt.Print("After delete: [")
	for i, u := range mapper.FindAll() {
		if i > 0 {
			fmt.Print(", ")
		}
		fmt.Print(u)
	}
	fmt.Println("]")
}

Related Patterns

• Repository — the repository pattern uses data mapper internally to translate between domain objects and database rows.
• Active Record — active record is a simpler alternative that combines domain and persistence in one class; use when domain logic is minimal.
• DTO — data transfer objects carry data between the mapper and the persistence layer without exposing domain internals.