AW Dev Rethought

βœ‚οΈ Perfection is achieved not when there is nothing more to add, but when there is nothing left to take away - Antoine de Saint-ExupΓ©ry

βš™οΈ Python Backend Patterns – πŸ—‚οΈ Repository Pattern


Description:

As backend applications grow, database operations can easily become scattered across routes, services, and other parts of the application.

The Repository Pattern solves this by introducing a dedicated layer responsible for accessing and modifying data.

Instead of application logic communicating directly with a database or another data source, it communicates through a repository.

Conceptually, the flow becomes:

Application / Service
        ↓
    Repository
        ↓
    Data Source

This keeps data-access concerns separate from the application's business logic.


Defining the Data Model

We'll use a simple User model:

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

For this example, we don't need an actual database. A Python dictionary will act as our in-memory data source so we can focus entirely on the pattern itself.


Creating the Repository

The UserRepository becomes the only component responsible for accessing user data.

class UserRepository:
    def __init__(self):
        self._users = {}

    def add(self, user):
        self._users[user.id] = user

    def get_by_id(self, user_id):
        return self._users.get(user_id)

    def get_all(self):
        return list(self._users.values())

Operations such as retrieving, storing, or deleting users remain inside this class.

If we later replace the dictionary with PostgreSQL, MongoDB, or an external API, the rest of the application doesn't need to contain that implementation logic.


Keeping Business Logic Separate

The service layer handles application rules rather than data storage.

class UserService:
    def __init__(self, repository):
        self.repository = repository

For example, when registering a user, the service can check whether that user already exists:

if self.repository.get_by_id(user_id):
    raise ValueError(f"User {user_id} already exists")

The service decides what should happen, while the repository handles how the data is accessed.

That's the important separation introduced by this pattern.


Using the Repository

We create the repository and inject it into the service:

repository = UserRepository()
service = UserService(repository)

Application operations then go through the service:

service.register_user(1, "Alice", "alice@example.com")
service.register_user(2, "Bob", "bob@example.com")

The application doesn't need to know whether the users are stored in a dictionary, database, or somewhere else.


Why Use the Repository Pattern?

The Repository Pattern gives backend applications a clearer separation between business logic and data access.

It can make the code:

  • easier to maintain
  • easier to test
  • less dependent on a particular database implementation
  • easier to extend when the persistence layer changes

It isn't necessary for every small application, but it becomes useful when data-access logic starts spreading across multiple parts of a backend.


Key Takeaways

  • A repository creates a dedicated abstraction for data access.
  • Business logic doesn't need to communicate directly with the underlying data source.
  • Data-access implementations can change with less impact on application logic.
  • Repositories also make it easier to substitute fake or in-memory implementations during testing.

Code Snippet:

# Import dataclass for defining the data model
from dataclasses import dataclass

# Import Optional for values that may not exist
from typing import Optional


# -----------------------------------------
# Step 1: Define the application data model
# -----------------------------------------
@dataclass
class User:
    id: int
    name: str
    email: str


# -----------------------------------------
# Step 2: Create the repository
# -----------------------------------------
class UserRepository:
    """Handles all user data access operations."""

    def __init__(self):
        # Use a dictionary as an in-memory data source
        self._users = {}

    def add(self, user: User) -> None:
        """Store a new user."""
        self._users[user.id] = user

    def get_by_id(self, user_id: int) -> Optional[User]:
        """Retrieve a user by ID."""
        return self._users.get(user_id)

    def get_all(self) -> list[User]:
        """Retrieve all users."""
        return list(self._users.values())

    def delete(self, user_id: int) -> bool:
        """Delete a user if it exists."""
        if user_id not in self._users:
            return False

        del self._users[user_id]
        return True


# -----------------------------------------
# Step 3: Create the service layer
# -----------------------------------------
class UserService:
    """Contains application business logic."""

    def __init__(self, repository: UserRepository):
        # Depend on the repository instead of the data source
        self.repository = repository

    def register_user(self, user_id: int, name: str, email: str) -> User:
        """Register a new user."""

        # Prevent duplicate user IDs
        if self.repository.get_by_id(user_id):
            raise ValueError(f"User {user_id} already exists")

        user = User(
            id=user_id,
            name=name,
            email=email,
        )

        self.repository.add(user)

        return user

    def find_user(self, user_id: int) -> Optional[User]:
        """Find an existing user."""
        return self.repository.get_by_id(user_id)

    def list_users(self) -> list[User]:
        """Return all registered users."""
        return self.repository.get_all()

    def remove_user(self, user_id: int) -> bool:
        """Remove an existing user."""
        return self.repository.delete(user_id)


# -----------------------------------------
# Step 4: Configure application components
# -----------------------------------------
repository = UserRepository()
service = UserService(repository)


# -----------------------------------------
# Step 5: Execute application operations
# -----------------------------------------
service.register_user(1, "Alice", "alice@example.com")
service.register_user(2, "Bob", "bob@example.com")
service.register_user(3, "Charlie", "charlie@example.com")

print("Registered Users:")

for user in service.list_users():
    print(f"{user.id}: {user.name} ({user.email})")


# Retrieve a user through the service
user = service.find_user(2)

if user:
    print(f"\nFound User: {user.name}")


# Delete a user through the service
service.remove_user(1)

print("\nUsers After Deletion:")

for user in service.list_users():
    print(f"{user.id}: {user.name} ({user.email})")

Link copied!

Comments

Add Your Comment

Comment Added!