Decoupling Domain from Persistence: Implementing the Repository Pattern in Python

Introduction and Context
In enterprise applications, one of the most common anti-patterns is mixing business rules (Domain Logic) with database queries (Persistence Logic). As your application grows, this practice leads to highly coupled, fragile, and hard-to-test code.

In his seminal book Patterns of Enterprise Application Architecture (PofEAA), Martin Fowler defines the Repository pattern as a mediator between the domain and data mapping layers, acting like an in-memory collection of domain objects.

How Does the Repository Pattern Work?
Instead of writing SQL queries or direct ORM calls inside your business services, you interact with an abstraction (an interface). The domain layer doesn’t know—and shouldn't care—whether the data comes from a PostgreSQL database, a local JSON file, or an external API. It simply requests or provides domain objects.

Real-World Case Study: Academic Tutoring Management System
Imagine we are building the backend for a university tutoring platform. We need to manage tutoring sessions without coupling our core business logic to a specific database engine.

1. The Domain Entity (Pure Business Logic) This class represents our core business domain. It is completely decoupled from any frameworks or database libraries.

Python

domain.py

from datetime import datetime
from dataclasses import dataclass

@dataclass
class TutoringSession:
id: str
student_id: str
tutor_id: str
date_time: datetime
subject: str
is_active: bool = True

def cancel_session(self):
    if self.date_time < datetime.now():
        raise ValueError("Cannot cancel a session that has already occurred.")
    self.is_active = False

1. The Repository Interface (Abstraction) We define the contract that any concrete persistence mechanism must implement.

Python

repository_interface.py

from abc import ABC, abstractmethod
from typing import List, Optional
from domain import TutoringSession

class ITutoringRepository(ABC):

@abstractmethod
def save(self, session: TutoringSession) -> None:
    """Saves or updates a tutoring session."""
    pass

@abstractmethod
def find_by_id(self, session_id: str) -> Optional[TutoringSession]:
    """Retrieves a session by its unique identifier."""
    pass

@abstractmethod
def find_active_by_tutor(self, tutor_id: str) -> List[TutoringSession]:
    """Retrieves all active tutoring sessions for a specific tutor."""
    pass

1. The Concrete Implementation (In-Memory for Infrastructure/Testing) While you would use SQLAlchemy or a database client for production, the pattern allows us to create an in-memory version instantaneously for blazing-fast software testing.

Python

in_memory_repository.py

from typing import List, Optional, Dict
from domain import TutoringSession
from repository_interface import ITutoringRepository

class InMemoryTutoringRepository(ITutoringRepository):
def init(self):
self._db: Dict[str, TutoringSession] = {}

def save(self, session: TutoringSession) -> None:
    self._db[session.id] = session

def find_by_id(self, session_id: str) -> Optional[TutoringSession]:
    return self._db.get(session_id)

def find_active_by_tutor(self, tutor_id: str) -> List[TutoringSession]:
    return [
        session for session in self._db.values()
        if session.tutor_id == tutor_id and session.is_active
    ]

1. The Service Layer (Use Case Execution) Notice how the service consumes the repository interface without knowing how or where the data is actually stored.

Python

service.py

from domain import TutoringSession
from repository_interface import ITutoringRepository

class TutoringService:
def init(self, repository: ITutoringRepository):
self.repository = repository

def schedule_new_session(self, session: TutoringSession) -> None:
    # Complementary business logic validation
    current_sessions = self.repository.find_active_by_tutor(session.tutor_id)
    if len(current_sessions) >= 5:
        raise Exception("The tutor has already reached the maximum limit of active sessions.")

    self.repository.save(session)

Key Takeaways and Benefits
Absolute Testability: You can unit test your TutoringService by passing the InMemoryTutoringRepository in milliseconds, without needing to spin up Docker containers or real databases.

Substitutability: If you ever decide to migrate from a relational SQL database to a NoSQL engine or Firebase, your core business layer (service.py and domain.py) remains completely untouched. You simply write a new repository class that implements ITutoringRepository.

🛠️ GitHub Repository Guide
To make your repository look highly professional to your professor and research group, structure your project cleanly like this:

Plaintext
enterprise-patterns-repository/
│
├── src/
│ ├── init.py
│ ├── domain.py
│ ├── repository_interface.py
│ ├── in_memory_repository.py
│ └── service.py
│
├── main.py # A small executable script demonstrating the workflow
├── README.md # Clear documentation on how to run the code
└── .gitignore
Inside your main.py, place a simple execution flow to showcase your pattern in action:

Python
from datetime import datetime
from src.domain import TutoringSession
from src.in_memory_repository import InMemoryTutoringRepository
from src.service import TutoringService

if name == "main":
# Initialize infrastructure and service layers
repo = InMemoryTutoringRepository()
service = TutoringService(repo)

# Create a domain object
new_session = TutoringSession(
    id="SES-001", 
    student_id="STUDENT-10", 
    tutor_id="TUTOR-05", 
    date_time=datetime(2026, 7, 1, 15, 0), 
    subject="Software Architecture Patterns"
)

# Execute the business use case
service.schedule_new_session(new_session)
print(f"Session {new_session.id} successfully saved to the Repository.")