The “Normal” ORM Way: Model Depends on ORM

These days it’s unlikely that your team members are hand-rolling their own SQL queries. Instead, you’re almost certainly using some kind of framework to generate SQL for you based on your model objects.

These frameworks are called object-relational mappers (ORMs) because they exist to bridge the conceptual gap between the world of objects and domain modeling, and the world of databases and relational algebra.

The most important thing an ORM gives us is persistence ignorance: the idea that our fancy domain model doesn’t need to know anything about how data are loaded or persisted. This helps to keep our domain clean of direct dependencies on particular databases technologies.³

But if you follow the typical SQLAlchemy tutorial, you’ll end up with something like this:

from sqlalchemy import Column, ForeignKey, Integer, String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import relationship

Base = declarative_base()

class Order(Base):
    id = Column(Integer, primary_key=True)

class OrderLine(Base):
    id = Column(Integer, primary_key=True)
    sku = Column(String(250))
    qty = Integer(String(250))
    order_id = Column(Integer, ForeignKey('order.id'))
    order = relationship(Order)

class Allocation(Base):
    ...

You don’t need to understand SQLAlchemy to see that our pristine model is now full of dependencies on the ORM, and is starting to look ugly as hell besides. Can we really say this model is ignorant of the database? How can it be separate from storage concerns when our model properties are directly coupled to database columns?

Inverting the Dependency: ORM Depends on Model

Well, thankfully, that’s not the only way to use SQLAlchemy. The alternative is to define your schema separately, and an explicit mapper for how to convert between the schema and our domain model, what SQLAlchemy calls a classical mapping.

from sqlalchemy.orm import mapper, relationship

import model  


metadata = MetaData()

order_lines = Table(  
    'order_lines', metadata,
    Column('id', Integer, primary_key=True, autoincrement=True),
    Column('sku', String(255)),
    Column('qty', Integer, nullable=False),
    Column('orderid', String(255)),
)

...

def start_mappers():
    lines_mapper = mapper(model.OrderLine, order_lines)  

The ORM imports (or “depends on” or “knows about”) the domain model, and not the other way around.

We define our database tables and columns using SQLAlchemy’s abstractions.⁴

And when we call the mapper function, SQLAlchemy does its magic to bind our domain model classes to the various tables we’ve defined.

The end result will be that, if we call start_mappers(), we will be able to easily load and save domain model instances from and to the database. But if we never call that function, then our domain model classes stay blissfully unaware of the database.

This gives us all the benefits of SQLAlchemy, including the ability to use alembic for migrations, and the ability to transparently query using our domain classes, as we’ll see.

When you’re first trying to build your ORM config, it can be useful to write some tests for it, as in Example 2-5:

def test_orderline_mapper_can_load_lines(session):  
    session.execute(  
        'INSERT INTO order_lines (orderid, sku, qty) VALUES '
        '("order1", "RED-CHAIR", 12),'
        '("order1", "RED-TABLE", 13),'
        '("order2", "BLUE-LIPSTICK", 14)'
    )
    expected = [
        model.OrderLine("order1", "RED-CHAIR", 12),
        model.OrderLine("order1", "RED-TABLE", 13),
        model.OrderLine("order2", "BLUE-LIPSTICK", 14),
    ]
    assert session.query(model.OrderLine).all() == expected


def test_orderline_mapper_can_save_lines(session):
    new_line = model.OrderLine("order1", "DECORATIVE-WIDGET", 12)
    session.add(new_line)
    session.commit()

    rows = list(session.execute('SELECT orderid, sku, qty FROM "order_lines"'))
    assert rows == [("order1", "DECORATIVE-WIDGET", 12)]

If you’ve not used pytest, the session argument to this test needs explaining. You don’t need to worry about the details of pytest or its fixtures for the purposes of this book, but the short explanation is that you can define common dependencies for your tests as “fixtures,” and pytest will inject them to the tests that need them by looking at their function arguments. In this case, it’s a SQLAlchemy database session.

You probably wouldn’t keep these tests around—as we’ll see shortly, once you’ve taken the step of inverting the dependency of ORM and Domain Model, it’s only a small additional step to implement an additional abstraction called the repository pattern, which will be easier to write tests against, and will provide a simple, common interface for faking out later in tests.

But we’ve already achieved our objective of inverting the traditional dependency: the domain model stays “pure” and free from infrastructure concerns. We could throw away SQLAlchemy and use a different ORM, or a totally different persistence system, and the domain model doesn’t need to change at all.

Depending on what you’re doing in your domain model, and especially if you stray far from the OO paradigm, you may find it increasingly hard to get the ORM to produce the exact behavior you need, and you may need to modify your domain model⁵. As so often with architectural decisions, there is a trade-off you’ll need to consider. As the Zen of Python says, “Practicality beats purity!”

At this point though, our API endpoint might look something like Example 2-6, and we could get it to work just fine.

@flask.route.gubbins
def allocate_endpoint():
    session = start_session()

    # extract order line from request
    line = OrderLine(
        request.json['orderid'],
        request.json['sku'],
        request.json['qty'],
    )

    # load all batches from the DB
    batches = session.query(Batch).all()

    # call our domain service
    allocate(line, batches)

    # save the allocation back to the database
    session.commit()

    return 201

The Repository in the Abstract

The simplest repository has just two methods: add to put a new item in the repository, and get to return a previously added item.⁶ We stick rigidly to using these methods for data access in our domain and our service layer. This self-imposed simplicity stops us from coupling our domain model to the database.

Here’s what an abstract base class (ABC) for our repository would look like:

class AbstractRepository(abc.ABC):

    @abc.abstractmethod  
    def add(self, batch: model.Batch):
        raise NotImplementedError  

    @abc.abstractmethod
    def get(self, reference) -> model.Batch:
        raise NotImplementedError

Python tip: @abc.abstractmethod is one of the only things that makes ABCs actually “work” in Python. Python will refuse to let you instantiate a class that does not implement all the abstractmethods defined in its parent class

raise NotImplementedError is nice but neither necessary nor sufficient. In fact, your abstract methods can have real behavior which subclasses can call out to, if you want.

What Is the Trade-Off?

You know they say economists know the price of everything and the value of nothing? Well, programmers know the benefits of everything and the tradeoffs of nothing.

Rich Hickey

Whenever we introduce an architectural pattern in this book, we’ll always be trying to ask: “What do we get for this? And what does it cost us?.”

Usually at the very least we’ll be introducing an extra layer of abstraction, and although we may hope it will be reducing complexity overall, it does add complexity locally, and it has a cost in terms raw numbers of moving parts and ongoing maintenance.

Repository pattern is probably one of the easiest choices in the book though, if you’ve already heading down the DDD and dependency inversion route. As far as our code is concerned, we’re really just swapping the SQLAlchemy abstraction (session.query(Batch)) for a different one (batches_repo.get) which we designed.

We will have to write a few lines of code in our repository class each time we add a new domain object that we want to retrieve, but in return we get a very simple abstraction over our storage layer, which we control. It would make it very easy to make fundamental changes to the way we store things (see Appendix C), and as we’ll see, it is very easy to fake out for unit tests.

In addition, Repository Pattern is so common in the DDD world that, if you do collaborate with programmers that have come to Python from the Java and C# worlds, they’re likely to recognize it. Figure 2-5 shows an illustration.

Figure 2-5. Repository pattern

[ditaa, repository_pattern_diagram]
  +-----------------------------+
  |      Application Layer      |
  +-----------------------------+
                 |^
                 ||          /------------------
                 ||----------|   Domain Model   |
                 ||          |      objects     |
                 ||          ------------------/
                 V|
  +------------------------------+
  |          Repository          |
  +------------------------------+
                 |
                 V
  +------------------------------+
  |        Database Layer        |
  +------------------------------+

As always, we start with a test. This would probably be classified as an integration test, since we’re checking that our code (the repository) is correctly integrated with the database; hence, the tests tend to mix raw SQL with calls and assertions on our own code.

def test_repository_can_save_a_batch(session):
    batch = model.Batch("batch1", "RUSTY-SOAPDISH", 100, eta=None)

    repo = repository.SqlAlchemyRepository(session)
    repo.add(batch)  
    session.commit()  

    rows = list(session.execute(
        'SELECT reference, sku, _purchased_quantity, eta FROM "batches"'  
    ))
    assert rows == [("batch1", "RUSTY-SOAPDISH", 100, None)]

repo.add() is the method under test here

We keep the .commit() outside of the repository, and make it the responsibility of the caller. There are pros and cons for this, some of our reasons will become clearer when we get to Chapter 6.

And we use the raw SQL to verify that the right data has been saved.

The next test involves retrieving batches and allocations so it’s more complex:

def insert_order_line(session):
    session.execute(  
        'INSERT INTO order_lines (orderid, sku, qty)'
        ' VALUES ("order1", "GENERIC-SOFA", 12)'
    )
    [[orderline_id]] = session.execute(
        'SELECT id FROM order_lines WHERE orderid=:orderid AND sku=:sku',
        dict(orderid="order1", sku="GENERIC-SOFA")
    )
    return orderline_id

def insert_batch(session, batch_id):  
    ...

def test_repository_can_retrieve_a_batch_with_allocations(session):
    orderline_id = insert_order_line(session)
    batch1_id = insert_batch(session, "batch1")
    insert_batch(session, "batch2")
    insert_allocation(session, orderline_id, batch1_id)  

    repo = repository.SqlAlchemyRepository(session)
    retrieved = repo.get("batch1")

    expected = model.Batch("batch1", "GENERIC-SOFA", 100, eta=None)
    assert retrieved == expected  # Batch.__eq__ only compares reference  
    assert retrieved.sku == expected.sku  
    assert retrieved._purchased_quantity == expected._purchased_quantity
    assert retrieved._allocations == {
        model.OrderLine("order1", "GENERIC-SOFA", 12),
    }  

This tests the read side, so the raw SQL is preparing data to be read by the repo.get()

We’ll spare you the details of insert_batch and insert_allocation, the point is to create a couple of different batches, and for the batch we’re interested in to have one existing order line allocated to it.

And that’s what we verify here. The first assert == checks that the types match, and that the reference is the same (because, if you remember, Batch is an Entity, and we have a custom eq for it).

So we also explicitly check on its major attributes, including ._allocations, which is a Python set of OrderLine Value Objects.

Whether or not you painstakingly write tests for every model is a judgement call. Once you have one class tested for create/modify/save, you might be happy to go on and do the others with a minimal roundtrip test, or even nothing at all, if they all follow a similar pattern. In our case, the ORM config that sets up the ._allocations set is a little complex, so it merited a specific test.

You end up with something like Example 2-11:

class SqlAlchemyRepository(AbstractRepository):

    def __init__(self, session):
        self.session = session

    def add(self, batch):
        self.session.add(batch)

    def get(self, reference):
        return self.session.query(model.Batch).filter_by(reference=reference).one()

    def list(self):
        return self.session.query(model.Batch).all()

And now our flask endpoint might look something like Example 2-12:

@flask.route.gubbins
def allocate_endpoint():
    batches = SqlAlchemyRepository.list()
    lines = [
        OrderLine(l['orderid'], l['sku'], l['qty'])
         for l in request.params...
    ]
    allocate(lines, batches)
    session.commit()
    return 201