SQLAlchemy Repository for Aggregate Persistence

Introduction

In our last session, we established a crucial architectural boundary. We created a "map" that connects our pure Product domain model to a database schema using SQLAlchemy's imperative mapping. This ensures our domain logic remains completely isolated from infrastructure concerns.

However, a map is only useful if you have a way to travel the roads it depicts. Today, we will build that "vehicle": a concrete repository implementation that uses our map to persist and retrieve Product aggregates from a real database.

Lesson Goal: By the end of this 60-minute lesson, you will be able to implement a SQLAlchemy-based repository that uses a Session to persist and retrieve aggregates. This is the "real" adapter for our persistence port, complementing the "fake" in-memory version we built for testing.

1. The SQLAlchemy Session: Your Workspace for Database Interaction

Before we write our repository, we need to understand its most important tool: the SQLAlchemy Session.

Think of the Session as a temporary workspace or a scratchpad. When you want to interact with the database, you open a session. You then load objects from the database into this workspace, make changes, or add new objects. The session keeps track of everything you've done. When you're finished, you tell the session to commit these changes to the database, making them permanent.

Key characteristics of the Session:

• It manages database connections.
• It provides the primary interface for all queries.
• It tracks changes to objects.
• It manages transaction boundaries (commit, rollback).

The video below explains how to create and use a session in modern SQLAlchemy.

This Session object is what our SQLAlchemy repository will use to do its job.

2. Implementing the SqlAlchemyProductRepository

Now, let's create our concrete repository. We will create a class SqlAlchemyProductRepository that implements the AbstractProductRepository interface we defined in a previous lesson.

A crucial design choice is that the repository will not create its own session. Instead, it will receive an active session in its constructor. This is a fundamental principle of the pattern: the repository performs data operations, but the calling code (an Application Service, as we'll see later) manages the transactional unit of work.

The "Architecture Patterns with Python" book (also known as Cosmic Python) provides the canonical implementation of this pattern, which we will follow.

Our Implementation

Let's apply this pattern to our Product aggregate. We'll create a new file for our repository implementation in the infrastructure layer.

# src/infrastructure/persistence/sqlalchemy/repositories/product_repository.py

import uuid
from sqlalchemy.orm import Session
from sqlalchemy import select

from src.domain.product.model import Product
from src.domain.product.repository import AbstractProductRepository

class SqlAlchemyProductRepository(AbstractProductRepository):
    def __init__(self, session: Session):
        self.session = session

    def add(self, product: Product):
        """
        Adds a new Product to the database.
        """
        self.session.add(product)

    def get(self, product_id: uuid.UUID) -> Product | None:
        """
        Retrieves a Product by its ID.
        Uses session.get(), the most efficient way to load by primary key.
        """
        return self.session.get(Product, product_id)

    def list(self) -> list[Product]:
        """
        Retrieves a list of all Products.
        """
        statement = select(Product)
        return list(self.session.scalars(statement).all())

Let's break down the methods:

• __init__(self, session: Session): We depend on an existing Session. This is dependency injection in action.
• add(self, product: Product): This is straightforward. We tell the session to start tracking our Product object. The session now considers this object "pending" for insertion.
• get(self, product_id: uuid.UUID): We use session.get(), which is the most direct way to fetch an object by its primary key. It's an improvement over the query().filter_by().one() syntax you saw in the Cosmic Python text, though both achieve a similar result.
• list(self): We use a select(Product) statement to indicate we want all columns from the table mapped to the Product class. session.scalars() executes this and returns the results as our Product domain objects.

3. Putting It All Together: A Usage Example

We have the mapping and the repository. How do we use them together?

Let's write a simple script to demonstrate the full lifecycle:

1. Set up the database connection (Engine).
2. Initialize our imperative mappers.
3. Create a session.
4. Instantiate our repository with the session.
5. Use the repository to add a new product.
6. Crucially, call session.commit() to save the changes.
7. Use the repository to get the product back.

# scripts/run_sqlalchemy_repo.py

import uuid
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

from src.domain.product.model import Product
from src.infrastructure.persistence.sqlalchemy.orm import start_mappers
from src.infrastructure.persistence.sqlalchemy.repositories.product_repository import SqlAlchemyProductRepository

# 1. Create an engine (using in-memory SQLite for this example)
engine = create_engine("sqlite:///:memory:")

# (From previous lesson) Create the table schema
from src.infrastructure.persistence.sqlalchemy.orm import metadata
metadata.create_all(engine)

# 2. Start the mappers
start_mappers()

# 3. Create a session factory
get_session = sessionmaker(bind=engine)

# --- Application Code Starts Here ---

# Create a new product
product_id = uuid.uuid4()
new_product = Product(id=product_id, sku="SKU123", name="Gaming Mouse", price=75.50)

# 4. Open a session and instantiate the repository
with get_session() as session:
    repo = SqlAlchemyProductRepository(session)

    # 5. Use the repository to add the product
    repo.add(new_product)

    # 6. Commit the transaction
    # The repository doesn't do this; the application logic does.
    session.commit()

# --- Verify the operation ---

# 7. In a new session, retrieve the product
with get_session() as session:
    repo = SqlAlchemyProductRepository(session)
    retrieved_product = repo.get(product_id)

    print(f"New product created: {new_product}")
    print(f"Retrieved product: {retrieved_product}")
    assert retrieved_product is not None
    assert retrieved_product.sku == "SKU123"

This script clearly demonstrates the separation of responsibilities. The repository handles the how of data access (session.add, session.get), while the application code manages the when of the transaction (session.commit).

4. Testing the Concrete Repository

While our InMemoryProductRepository is perfect for fast unit tests of our domain logic, we also need to ensure our SqlAlchemyProductRepository correctly integrates with the database. These are called integration tests.

We can write a test using pytest that uses a real (but temporary) database to verify that our repository can save and load data correctly. This is the same principle you saw in the Cosmic Python reading.

Here's what an integration test for our add and get methods would look like:

# tests/integration/test_product_repository.py

import pytest
import uuid
from src.domain.product.model import Product
from src.infrastructure.persistence.sqlalchemy.repositories.product_repository import SqlAlchemyProductRepository

def test_repository_can_add_and_get_a_product(session):
    # `session` is a pytest fixture that provides a clean database session for each test
    repo = SqlAlchemyProductRepository(session)
    
    product_id = uuid.uuid4()
    new_product = Product(id=product_id, sku="TEST-SKU", name="Test Product", price=99.99)
    
    repo.add(new_product)
    session.commit() # Commit to save the product

    retrieved_product = repo.get(product_id)
    
    assert retrieved_product is not None
    assert retrieved_product.sku == "TEST-SKU"
    assert retrieved_product.name == "Test Product"

This test gives us confidence that our mapping and repository implementation work together as expected.

Conclusion

Congratulations! You have now built a complete, production-ready persistence layer for your Product aggregate. You have a pure domain model, a clean mapping definition, and a concrete repository that encapsulates all the database interaction logic.

Key Takeaways:

• The SQLAlchemy Session is the central object for database operations, acting as a transactional workspace.
• A SQLAlchemy-based repository implements the domain's abstract repository interface. It takes a Session as a dependency and uses it to execute queries.
• Modern SQLAlchemy provides efficient methods like session.get() for primary key lookups and session.scalars(select(...)) for queries.
• The repository is responsible for data access logic, but not for transaction management. session.commit() and session.rollback() are handled by the code that uses the repository.

Preview of the Next Lesson:

We've repeatedly said that "the caller" or "the application" is responsible for managing the session and transactions. But what does that code look like? In our next module, we will dive into Application Services and the Unit of Work pattern, which provide a robust and elegant structure for orchestrating use cases and managing transactions that span one or more repository operations.