Mapping Aggregate Roots to SQLAlchemy Models

Introduction

In our previous lesson, we built our first repository implementation: a simple InMemoryProductRepository. This "fake" repository is a crucial tool for fast, isolated unit testing, acting as a test-friendly "Adapter" for our domain's persistence "Port."

Now, it's time to build the "real" adapter—one that connects to a persistent database. The first and most critical step is to define how our domain objects, specifically our Product Aggregate Root, are represented in a database. We need to create a map that translates our Python object's attributes to the columns of a database table.

Lesson Goal: By the end of this 60-minute lesson, you will be able to set up a basic SQLAlchemy model to map an Aggregate Root to a database schema. We will use SQLAlchemy, a powerful Object-Relational Mapper (ORM), to create this link between our code and the database.

1. What is an ORM? Meet SQLAlchemy

An Object-Relational Mapper (ORM) is a library that acts as a bridge, automating the transfer of data between object-oriented programs (like our Python application) and relational databases (like PostgreSQL or SQLite). Instead of writing raw SQL queries like INSERT INTO ... or SELECT * FROM ..., we can manipulate Python objects, and the ORM handles the SQL translation for us.

SQLAlchemy is the de facto standard ORM in the Python ecosystem. To get a quick sense of what it does, let's watch a brief introduction.

As the video explains, the ORM lets us interact with the database using Python objects. This is perfect for DDD, as it allows us to work primarily with our rich domain models (Product, Order, etc.) and delegate the persistence details.

2. Two Paths to Mapping: Declarative vs. Imperative

SQLAlchemy offers two main ways to define the mapping between your Python classes and database tables. The choice between them has significant architectural implications, especially in a DDD context.

Path A: The Declarative Approach (Coupled)

This is the most common approach you'll see in tutorials and simpler applications. The mapping information is placed directly inside the model class using decorators or special base classes.

Let's look at the official SQLAlchemy documentation to see how this works.

In the Declarative style, the User class is tightly coupled to SQLAlchemy. It knows it's being mapped to a table named user_account and knows the specific types and constraints of its columns. While convenient, this violates a core principle of DDD: the domain model should be pure and ignorant of infrastructure concerns like the database.

Path B: The Imperative Approach (Decoupled)

To maintain the purity of our domain model, we can use SQLAlchemy's Imperative Mapping (also known as Classical Mapping). With this approach, our domain model remains a "plain" Python object. The mapping is defined separately, in the infrastructure layer, where it belongs.

This creates a clean separation:

• Domain Layer: Contains the Product class, with only business logic. It has no SQLAlchemy imports.
• Infrastructure Layer: Contains the mapping code that tells SQLAlchemy, "take that Product class and map it to this database table."

This is the approach we will take. It perfectly aligns with the Ports and Adapters architecture.

The following resource shows exactly how to implement this separation.

This is the pattern we will follow. It keeps our domain model completely clean, just as we want.

3. Implementing the Mapping for our Product Aggregate

Let's apply the Imperative Mapping pattern to our Product aggregate root.

First, recall our Product model. It's a simple dataclass and lives in the domain layer. It has no knowledge of databases.

# src/domain/product/model.py

import uuid
from dataclasses import dataclass, field

@dataclass
class Product:
    id: uuid.UUID = field(default_factory=uuid.uuid4)
    sku: str
    name: str
    price: float

Now, we'll create a new file in our infrastructure layer to define the database mapping.

Step 1: Define the SQLAlchemy Table

We'll create a Table object that describes the products table in our database.

Step 2: Define the Mapper Function

We'll create a function, let's call it start_mappers(), that uses map_imperatively to link our Product class to the products_table.

Here is the complete code for our new mapping file:

# src/infrastructure/persistence/sqlalchemy/orm.py

from sqlalchemy import (
    Table,
    MetaData,
    Column,
    String,
    Float,
)
from sqlalchemy.dialects.postgresql import UUID
from sqlalchemy.orm import registry

from src.domain.product.model import Product

# Create a metadata object. This is a registry of tables and their associated schema.
metadata = MetaData()

# Create a registry that will hold our mappers.
mapper_registry = registry()

# Define the 'products' table. This is pure database schema definition.
products_table = Table(
    "products",
    metadata,
    Column("id", UUID(as_uuid=True), primary_key=True),
    Column("sku", String(255), nullable=False, unique=True),
    Column("name", String(255), nullable=False),
    Column("price", Float, nullable=False),
)

def start_mappers():
    """
    This function tells SQLAlchemy how to map our domain models to the database tables.
    """
    # Clear any existing mappers to ensure a clean start, useful for testing.
    mapper_registry.dispose()

    # Map the Product domain model to the products_table.
    mapper_registry.map_imperatively(Product, products_table)

Look at the clean separation! The orm.py file lives in infrastructure and handles all the database-specific definitions. Our Product model in domain remains completely untouched and unaware of SQLAlchemy.

4. Generating the Database Schema

We have defined the what (the Product class) and the how (the products_table and the imperative mapping). Now, let's use this metadata to generate the actual CREATE TABLE SQL statement and create our database schema.

This involves two final steps:

1. Create an Engine: The Engine is a SQLAlchemy object that manages connections to a specific database (e.g., a local SQLite file).
2. Emit DDL: Use the metadata object we defined to create all tables in the database.

We can write a small, standalone script to initialize our database.

# scripts/create_db.py

from sqlalchemy import create_engine
from src.infrastructure.persistence.sqlalchemy.orm import metadata

# We'll use SQLite for simplicity. This will create a file named 'app.db'
# The 'echo=True' argument will print all the SQL that SQLAlchemy generates.
engine = create_engine("sqlite:///app.db", echo=True)

def create_database():
    print("Creating database schema...")
    # This command connects to the database via the engine and creates all tables
    # defined in our metadata object.
    metadata.create_all(engine)
    print("Database schema created.")

if __name__ == "__main__":
    create_database()

If you were to run this script (python scripts/create_db.py), you would see output like this, which is SQLAlchemy generating and executing the SQL for you:

INFO:sqlalchemy.engine.Engine:BEGIN (implicit)
INFO:sqlalchemy.engine.Engine:PRAGMA main.table_info("products")
INFO:sqlalchemy.,"engine.Engine:()
INFO:sqlalchemy.engine.Engine:CREATE TABLE products (
    id VARCHAR NOT NULL,
    sku VARCHAR(255) NOT NULL,
    name VARCHAR(255) NOT NULL,
    price FLOAT NOT NULL,
    PRIMARY KEY (id),
    UNIQUE (sku)
)
INFO:sqlalchemy.engine.Engine:()
INFO:sqlalchemy.engine.Engine:COMMIT

We have successfully mapped our pure domain object to a database schema!

Conclusion

Today we took a critical step in building our persistence layer. We connected our abstract domain model to a concrete database structure, all while maintaining the strict separation of concerns that is central to DDD.

Key Takeaways:

• SQLAlchemy is an ORM that bridges the gap between Python objects and relational databases.
• For DDD, Imperative (or Classical) Mapping is the preferred approach because it keeps the domain model free of persistence-related code.
• The mapping process involves:
  1. Defining a Table object in the infrastructure layer to represent the database schema.
  2. Using mapper_registry.map_imperatively() to link the domain class to the Table object.
• This mapping metadata, combined with a SQLAlchemy Engine, is used to generate the database schema with metadata.create_all().

Preview of the Next Lesson:

We have the map, but we don't have a vehicle to travel the road yet. In the next lesson, we will build that vehicle. We will implement a SQLAlchemy-based repository that uses this mapping and a SQLAlchemy Session to persist and retrieve our Product aggregates from the database.