4  DAO (Data Access Object)

4.1 1 What problem does DAO solve?

Goal: decouple business / presentation logic from the details of talking to a data store (SQL, NoSQL, APIs, in-memory lists, …).

Without DAO, every place that needs data would execute SQL, open files, or make HTTP calls directly. With DAO you:

• hide the how of persistence behind a small, stable interface;
• can swap the storage mechanism (SQLite → PostgreSQL → REST) with minimal impact;
• make unit testing easy by injecting a fake/in-memory DAO.

---

4.2 2 Roles in the pattern

Role	Responsibility
Entity / DTO	Plain data structure (e.g. Book)
DAO interface	Declares CRUD operations (create, get, update, delete, …)
DAO implementation(s)	Concrete classes that do the real persistence work
Service / business layer	Uses the DAO; knows nothing about how data is stored

---

4.3 3 Tiny UML sketch

+-----------------+    uses    +-------------------+
|  BookService    | ---------> |  BookDAO (iface)  |
+-----------------+            +-------------------+
                                   ^         ^
             implements            |         | implements
        +-------------------+  +----------------------+
        | InMemoryBookDAO   |  |  SQLiteBookDAO       |
        +-------------------+  +----------------------+

---

4.4 4 Minimal Python example

from __future__ import annotations
from typing import Protocol, List, Optional
import sqlite3
import contextlib
from dataclasses import dataclass

# ---------- Entity ----------------------------------------------------------
@dataclass
class Book:
    id: int | None  # None until stored
    title: str
    author: str
    year: int

# ---------- DAO interface ---------------------------------------------------
class BookDAO(Protocol):
    def create(self, book: Book) -> Book: ...
    def get(self, book_id: int) -> Optional[Book]: ...
    def list_all(self) -> List[Book]: ...
    def delete(self, book_id: int) -> None: ...

# ---------- In-memory implementation (good for tests) -----------------------
class InMemoryBookDAO(BookDAO):
    def __init__(self) -> None:
        self._db: dict[int, Book] = {}
        self._next_id = 1

    def create(self, book: Book) -> Book:
        book.id = self._next_id
        self._db[self._next_id] = book
        self._next_id += 1
        return book

    def get(self, book_id: int) -> Optional[Book]:
        return self._db.get(book_id)

    def list_all(self) -> List[Book]:
        return list(self._db.values())

    def delete(self, book_id: int) -> None:
        self._db.pop(book_id, None)

# ---------- SQLite implementation (real persistence) ------------------------
class SQLiteBookDAO(BookDAO):
    def __init__(self, db_path: str = ":memory:") -> None:
        self.conn = sqlite3.connect(db_path, check_same_thread=False)
        with self.conn:
            self.conn.execute("""
                CREATE TABLE IF NOT EXISTS books (
                    id INTEGER PRIMARY KEY AUTOINCREMENT,
                    title TEXT NOT NULL,
                    author TEXT NOT NULL,
                    year INTEGER NOT NULL
                )
            """)

    def create(self, book: Book) -> Book:
        with self.conn:
            cur = self.conn.execute(
                "INSERT INTO books(title,author,year) VALUES (?,?,?)",
                (book.title, book.author, book.year)
            )
            book.id = cur.lastrowid
        return book

    def get(self, book_id: int) -> Optional[Book]:
        row = self.conn.execute(
            "SELECT id,title,author,year FROM books WHERE id = ?",
            (book_id,)
        ).fetchone()
        return Book(*row) if row else None

    def list_all(self) -> List[Book]:
        rows = self.conn.execute("SELECT id,title,author,year FROM books").fetchall()
        return [Book(*r) for r in rows]

    def delete(self, book_id: int) -> None:
        with self.conn:
            self.conn.execute("DELETE FROM books WHERE id = ?", (book_id,))

# ---------- Business/service layer ------------------------------------------
class BookService:
    def __init__(self, dao: BookDAO) -> None:
        self.dao = dao                              # <- Dependency Injection!

    def add_new_book(self, title: str, author: str, year: int) -> Book:
        return self.dao.create(Book(None, title, author, year))

    def show_catalogue(self) -> None:
        for book in self.dao.list_all():
            print(f"{book.id}: {book.title} ({book.year}) by {book.author}")

# ---------- Demo -------------------------------------------------------------
if __name__ == "__main__":
    # swap DAO here without changing the service
    service = BookService(dao=SQLiteBookDAO("library.db"))
    # service = BookService(dao=InMemoryBookDAO())   # <- easy testing

    service.add_new_book("Clean Code", "Robert C. Martin", 2008)
    service.add_new_book("Design Patterns", "GoF", 1994)
    service.show_catalogue()

4.4.1 Why this is DAO

• BookService asks only for BookDAO; it does not know where books live.
• We can inject either InMemoryBookDAO (unit tests) or SQLiteBookDAO (production).
• Switching to a completely different persistence (e.g. Mongo, REST API) means adding another DAO implementation without touching BookService.

---

4.5 5 Key take-aways

1. Single Responsibility & Open/Closed – storage logic lives in one place, business logic remains clean, and you can add new data sources without modifying client code.
2. Testability – inject a lightweight in-memory DAO to speed up unit tests.
3. Maintainability – SQL changes, connection-pooling concerns, retry policies, etc. are hidden behind the DAO interface.

DAO is especially handy in Python projects once your data access code starts creeping into view functions, models, or Jupyter notebooks—extract it early, and future-you will thank present-you!