[Architecture Patterns] Writing the Repository Pattern in Python

This article was written using AI.

1. Why the Repository Pattern Is Needed

The Traditional Problem: Mixing Business Logic with Data Access

Without the repository pattern, data access code tends to be written directly into business logic.

# ❌ Without the pattern
import sqlite3

def register_user(user_id: int, name: str, email: str) -> None:
    conn = sqlite3.connect("app.db")
    row = conn.execute("SELECT id FROM users WHERE id = ?", (user_id,)).fetchone()
    if row:
        raise ValueError(f"User {user_id} already exists")
    conn.execute("INSERT INTO users VALUES (?, ?, ?)", (user_id, name, email))
    conn.commit()

This code has three problems:

1. Fragile to change If you want to switch from SQLite to PostgreSQL, you have to rewrite register_user directly. Data access concerns end up bleeding into business logic.

2. Difficult to test Because a DB connection is created directly inside the function, a real DB is required even during testing. Every test requires DB setup and teardown, making tests slow and unreliable.

3. Duplicate SQL everywhere As the application grows, similar SQL statements get duplicated across the service layer, controller layer, and elsewhere. This becomes a breeding ground for missed fixes and bugs.

The Solution: The Repository Pattern

The repository pattern is a design pattern that separates "how data is stored and retrieved" from business logic and encapsulates it in a dedicated class (the repository).

The key is a two-level separation:

• An abstract repository (interface) is placed in the domain layer to define only "what can be done"
• A concrete implementation (SQLite, in-memory, etc.) is isolated in the infrastructure layer to encapsulate "how it is done"

# ✅ With the repository pattern
class UserService:
    def __init__(self, repo: UserRepository):  # depends on the abstraction
        self.repo = repo

    def register(self, user_id: int, name: str, email: str) -> None:
        if self.repo.find_by_id(user_id):
            raise ValueError(f"User {user_id} already exists")
        self.repo.save(User(id=user_id, name=name, email=email))

UserService only depends on the UserRepository interface. Because it doesn't need to know whether the backend is SQLite or in-memory, the following becomes possible:

• Even if you switch to a different DB type, you don't need to touch UserService at all
• During testing, simply swap in InMemoryUserRepository to run fast tests without a DB
• Since data access code is consolidated in the repository class, SQL duplication doesn't occur

---

2. Understanding the Overall Structure

Three Layers and Their Roles

The repository pattern divides an application into three layers.

Layer	Role	May Depend On
domain	Defines business concepts. Houses entities and abstract repositories.	Nothing (innermost layer)
service	Implements business logic. Handles data through abstract repositories.	domain layer only
infrastructure	Responsible for actual data storage and retrieval. Encapsulates interaction with DBs and APIs.	domain layer only

infrastructure knows about domain, but domain does not know about infrastructure. This unidirectional dependency is the core of the pattern. If the inner layer (domain) were to depend on the outer layer (infrastructure), any change to the DB would require modifying the business logic.

                  ┌─────────────────────────┐
                  │       domain layer       │
                  │  User (entity)           │
                  │  UserRepository (abstract)│
                  └────────────┬────────────┘
                               │ depends on
          ┌────────────────────┼───────────────────┐
          │                   │                   │
          ▼                   ▼                   ▼
  ┌───────────────┐  ┌────────────────┐  ┌───────────────────┐
  │ service layer │  │infrastructure  │  │     main.py        │
  │  UserService  │  │ InMemory impl  │  │ only this knows    │
  │               │  │ SQLite impl    │  │ which impl to use  │
  └───────────────┘  └────────────────┘  └───────────────────┘

Directory Structure

Separating by layer makes the dependency relationships visible in the folder structure.

my_app/
├── domain/
│   ├── user.py                        # entity
│   └── user_repository.py             # abstract repository
├── infrastructure/
│   ├── in_memory_user_repository.py   # in-memory implementation (for testing/development)
│   └── sqlite_user_repository.py      # SQLite implementation (for production)
├── service/
│   └── user_service.py                # business logic
└── main.py                            # entry point (implementation selection)

---

3. Reading the Implementation

domain Layer: Defining Business Concepts

domain/user.py — Entity

The central data structure of the application. It represents the business concept of a "user" in pure terms, without any dependency on database or communication concerns. This is the innermost layer, with no dependencies on any other module.

from dataclasses import dataclass

@dataclass
class User:
    id: int
    name: str
    email: str

domain/user_repository.py — Abstract Repository

Defines only the interface (contract) for "how users are stored and retrieved." No concrete implementation (SQL, API, etc.) appears here whatsoever. By placing it in the domain layer, business logic can be written with the assumption that "any data source can be handled uniformly."

from abc import ABC, abstractmethod
from domain.user import User

class UserRepository(ABC):
    @abstractmethod
    def find_by_id(self, user_id: int) -> User | None:
        pass

    @abstractmethod
    def save(self, user: User) -> None:
        pass

    @abstractmethod
    def find_all(self) -> list[User]:
        pass

infrastructure Layer: Encapsulating Implementation Details

The same UserRepository interface is implemented using different technologies. Either one can be used interchangeably from the service layer.

infrastructure/in_memory_user_repository.py — In-Memory Implementation

A simple implementation that uses a dictionary as storage. No DB setup is required, making it valuable for testing and early development. As long as the UserRepository contract is satisfied, the internal implementation can be anything.

from domain.user import User
from domain.user_repository import UserRepository

class InMemoryUserRepository(UserRepository):
    def __init__(self):
        self._store: dict[int, User] = {}

    def find_by_id(self, user_id: int) -> User | None:
        return self._store.get(user_id)

    def save(self, user: User) -> None:
        self._store[user.id] = user

    def find_all(self) -> list[User]:
        return list(self._store.values())

infrastructure/sqlite_user_repository.py — SQLite Implementation

The same interface is now implemented with SQLite. The key point is that "infrastructure details" like SQL statements and connection management are all encapsulated within this class alone. If you later switch to PostgreSQL or MongoDB, it's just a matter of swapping out this file.

import sqlite3
from domain.user import User
from domain.user_repository import UserRepository

class SQLiteUserRepository(UserRepository):
    def __init__(self, db_path: str):
        self.conn = sqlite3.connect(db_path)
        self.conn.execute(
            "CREATE TABLE IF NOT EXISTS users (id INTEGER PRIMARY KEY, name TEXT, email TEXT)"
        )

    def find_by_id(self, user_id: int) -> User | None:
        row = self.conn.execute(
            "SELECT id, name, email FROM users WHERE id = ?", (user_id,)
        ).fetchone()
        return User(*row) if row else None

    def save(self, user: User) -> None:
        self.conn.execute(
            "INSERT OR REPLACE INTO users VALUES (?, ?, ?)",
            (user.id, user.name, user.email)
        )
        self.conn.commit()

    def find_all(self) -> list[User]:
        rows = self.conn.execute("SELECT id, name, email FROM users").fetchall()
        return [User(*row) for row in rows]

service Layer: Writing Business Logic

service/user_service.py

Application-specific rules — such as checking for duplicate users at registration — are written here. Because it only depends on the UserRepository abstraction, it is completely unaware of whether the backend is SQLite or in-memory. This is the greatest benefit of the repository pattern.

from domain.user import User
from domain.user_repository import UserRepository

class UserService:
    def __init__(self, repo: UserRepository):
        self.repo = repo

    def register(self, user_id: int, name: str, email: str) -> None:
        if self.repo.find_by_id(user_id):
            raise ValueError(f"User {user_id} already exists")
        self.repo.save(User(id=user_id, name=name, email=email))

    def get_user(self, user_id: int) -> User:
        user = self.repo.find_by_id(user_id)
        if not user:
            raise ValueError(f"User {user_id} not found")
        return user

---

4. Assembling It — main.py

This is the one place that decides which implementation to use. By swapping out just the repo = ... line, you can switch operating environments. Since UserService itself doesn't need to know about this choice, you can change the environment without touching the business logic at all.

from infrastructure.in_memory_user_repository import InMemoryUserRepository
from infrastructure.sqlite_user_repository import SQLiteUserRepository
from service.user_service import UserService

# For testing/development
repo = InMemoryUserRepository()

# For production (just swap one line)
# repo = SQLiteUserRepository("prod.db")

service = UserService(repo=repo)

service.register(1, "Taro Tanaka", "tanaka@example.com")
service.register(2, "Hanako Suzuki", "suzuki@example.com")

user = service.get_user(1)
print(user)

all_users = repo.find_all()
print(all_users)

Only main.py knows the whole picture. Neither the service layer nor the infrastructure layer needs to know about each other directly — they collaborate through the abstractions in the domain layer.

---

5. Summary

Here is a summary of the three problems the repository pattern solves and the means to solve them.

Problem	Solution
Changing the DB requires changes to business logic	By depending on the abstract repository, swapping implementations is isolated from business logic
Tests require a real DB and are slow and unreliable	Simply swap in InMemoryUserRepository to enable DB-free testing
SQL is scattered everywhere and hard to maintain	Data access code is consolidated in the repository class

The core of the design is keeping the direction of dependencies unidirectional. The domain layer doesn't need to know anything; only the infrastructure layer knows about the domain layer. By following this principle, you can protect the inside of your application (business logic) from the concerns of the outside (DB type, communication method) indefinitely.

---

Supplement: How Does main.py Change When Extended to a Web App?

In this sample, main.py handles both assembling instances and calling services, but when extended to a web app, the roles are separated.

Before (this sample)

my_app/
├── domain/
├── infrastructure/
├── service/
└── main.py   ← also handles direct service calls

After (web app)

my_app/
├── domain/
├── infrastructure/
├── service/
├── controller/
│   └── user_controller.py   ← receives HTTP requests and calls Service
├── dependencies.py           ← centralizes DI assembly
└── main.py                  ← server startup only

The controller is a dedicated layer that receives HTTP requests, delegates processing to the service, and returns a response. main.py focuses solely on starting the server and no longer participates in request handling.

DI assembly is centralized in dependencies.py. Even as endpoints increase, DB connection configuration changes are contained to this one place. With FastAPI, using Depends() for dependency injection causes get_user_service() to be called on each request, and the returned UserService instance is automatically injected into the controller's arguments.

dependencies.py

def get_user_service() -> UserService:
    repo = SQLiteUserRepository("prod.db")
    return UserService(repo=repo)

controller/user_controller.py (FastAPI example)

@router.post("/users")
def create_user(body: CreateUserRequest, service: UserService = Depends(get_user_service)):
    service.register(body.id, body.name, body.email)

main.py

app = FastAPI()
app.include_router(user_controller.router)

The three layers of domain / service / infrastructure remain unchanged. Extending to a web app is simply a matter of adding a controller layer to main.py.