Back to our allocations project! Before: we drive our app by talking to Repositories and the Domain Model shows where we got to at the end of the Repository chapter:
In this chapter, we discuss the difference between orchestration logic, business logic, and interfacing code, and we introduce the Service Layer pattern to take care of orchestrating our workflows and defining the use cases of our system.
We’ll also discuss testing: by combining the Service Layer with our Repository abstraction over the database, we’re able to write fast tests, not just of our domain model, but the entire workflow for a use case.
The Service Layer will become the main way into our app shows where we’re aiming for: we’re going to
add a Flask API that will talk to the Service Layer, which will serve as the
entrypoint to our Domain Model. By making the service layer depend on the
AbstractRepository, we’ll be able to unit test it using FakeRepository, and
then run it in real life using SqlAlchemyRepository.
|
Note
|
In our diagrams, we are using the convention that new components are highlighted with bold text/lines (and yellow/orange color, if you’re reading a digital version) |
|
Tip
|
You can find our code for this chapter at github.com/cosmicpython/code/tree/chapter_04_service_layer. git clone https://github.com/cosmicpython/code.git && cd code git checkout chapter_04_service_layer # or, if you want to code along, checkout the previous chapter: git checkout chapter_02_repository |
Like any good agile team, we’re hustling to try and get an MVP out and in front of the users to start gathering feedback. We have the core of our domain model and the domain service we need to allocate orders, and we have the Repository interface for permanent storage.
Let’s try and plug all the moving parts together as quickly as we can, and then refactor towards a cleaner architecture. Here’s our plan:
-
Use Flask to put an API endpoint in front of our
allocatedomain service. Wire up the database session and our repository. Test it with an end-to-end test and some quick and dirty SQL to prepare test data. -
Refactor out a Service Layer to serve as an abstraction to capture the use case, and sit between Flask and our Domain Model. Build some service-layer tests and show how they can use the
FakeRepository. -
Experiment with different types of parameters for our service layer functions; show that using primitive data types allows the service-layer’s clients (our tests and our flask API) to be decoupled from the model layer.
No-one is interested in getting into a long terminology debate about what counts as an E2E test versus a functional test versus an acceptance test versus an integration test versus unit tests. Different projects need different combinations of tests, and we’ve seen perfectly successful projects just split things into "fast tests" and "slow tests."
For now we want to write one or maybe two tests that are going to exercise a "real" API endpoint (using HTTP) and talk to a real database. Let’s call them end-to-end tests because it’s one of the most self-explanatory names.
A first API test (test_api.py) shows a first cut:
@pytest.mark.usefixtures('restart_api')
def test_api_returns_allocation(add_stock):
sku, othersku = random_sku(), random_sku('other') #(1)
batch1, batch2, batch3 = random_batchref(1), random_batchref(2), random_batchref(3)
add_stock([ #(2)
(batch1, sku, 100, '2011-01-02'),
(batch2, sku, 100, '2011-01-01'),
(batch3, othersku, 100, None),
])
data = {'orderid': random_orderid(), 'sku': sku, 'qty': 3}
url = config.get_api_url() #(3)
r = requests.post(f'{url}/allocate', json=data)
assert r.status_code == 201
assert r.json()['batchref'] == batch2-
random_sku(),random_batchref()etc are little helper functions that add generate some randomized characters using theuuidmodule. Because we’re running against an actual database now, this is one way to prevent different tests and runs from interfering with each other. -
add_stockis a helper fixture that just hides away the details of manually inserting rows into the database using SQL. We’ll find a nicer way of doing this later in the chapter. -
config.py is a module for getting configuration information. Again, this is an unimportant detail, and everyone has different ways of solving these problems, but if you’re curious, you can find out more in [appendix_project_structure].
Everyone solves these problems in different ways, but you’re going to need some way of spinning up Flask, possibly in a container, and also talking to a postgres database. If you want to see how we did it, check out [appendix_project_structure].
Implementing things in the most obvious way, you might get something like this:
from flask import Flask, jsonify, request
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
import config
import model
import orm
import repository
orm.start_mappers()
get_session = sessionmaker(bind=create_engine(config.get_postgres_uri()))
app = Flask(__name__)
@app.route("/allocate", methods=['POST'])
def allocate_endpoint():
session = get_session()
batches = repository.SqlAlchemyRepository(session).list()
line = model.OrderLine(
request.json['orderid'],
request.json['sku'],
request.json['qty'],
)
batchref = model.allocate(line, batches)
return jsonify({'batchref': batchref}), 201So far so good. No need for too much more of your "architecture astronaut" nonsense, Bob and Harry, you may be thinking.
But hang on a minute—there’s no commit. We’re not actually saving our allocation to the database. Now we need a second test, either one that will inspect the database state after (not very black-boxey), or maybe one that checks that we can’t allocate a second line if a first should have already depleted the batch:
@pytest.mark.usefixtures('restart_api')
def test_allocations_are_persisted(add_stock):
sku = random_sku()
batch1, batch2 = random_batchref(1), random_batchref(2)
order1, order2 = random_orderid(1), random_orderid(2)
add_stock([
(batch1, sku, 10, '2011-01-01'),
(batch2, sku, 10, '2011-01-02'),
])
line1 = {'orderid': order1, 'sku': sku, 'qty': 10}
line2 = {'orderid': order2, 'sku': sku, 'qty': 10}
url = config.get_api_url()
# first order uses up all stock in batch 1
r = requests.post(f'{url}/allocate', json=line1)
assert r.status_code == 201
assert r.json()['batchref'] == batch1
# second order should go to batch 2
r = requests.post(f'{url}/allocate', json=line2)
assert r.status_code == 201
assert r.json()['batchref'] == batch2Not quite so lovely, but that will force us to get a commit in.
If we keep going like this though, things are going to get uglier and uglier.
Supposing we want to add a bit of error-handling. What if the domain raises an error, for a sku that’s out of stock? Or what about a sku that doesn’t even exist? That’s not something the domain even knows about, nor should it. It’s more of a sanity-check that we should implement at the database layer, before we even invoke the domain service.
Now we’re looking at two more end-to-end tests:
@pytest.mark.usefixtures('restart_api')
def test_400_message_for_out_of_stock(add_stock): #(1)
sku, smalL_batch, large_order = random_sku(), random_batchref(), random_orderid()
add_stock([
(smalL_batch, sku, 10, '2011-01-01'),
])
data = {'orderid': large_order, 'sku': sku, 'qty': 20}
url = config.get_api_url()
r = requests.post(f'{url}/allocate', json=data)
assert r.status_code == 400
assert r.json()['message'] == f'Out of stock for sku {sku}'
@pytest.mark.usefixtures('restart_api')
def test_400_message_for_invalid_sku(): #(2)
unknown_sku, orderid = random_sku(), random_orderid()
data = {'orderid': orderid, 'sku': unknown_sku, 'qty': 20}
url = config.get_api_url()
r = requests.post(f'{url}/allocate', json=data)
assert r.status_code == 400
assert r.json()['message'] == f'Invalid sku {unknown_sku}'-
In the first test we’re trying to allocate more units than we have in stock
-
In the second, the sku just doesn’t exist (because we never called
add_stock), so it’s invalid as far as our app is concerned.
And, sure we could implement it in the Flask app too:
def is_valid_sku(sku, batches):
return sku in {b.sku for b in batches}
@app.route("/allocate", methods=['POST'])
def allocate_endpoint():
session = get_session()
batches = repository.SqlAlchemyRepository(session).list()
line = model.OrderLine(
request.json['orderid'],
request.json['sku'],
request.json['qty'],
)
if not is_valid_sku(line.sku, batches):
return jsonify({'message': f'Invalid sku {line.sku}'}), 400
try:
batchref = model.allocate(line, batches)
except model.OutOfStock as e:
return jsonify({'message': str(e)}), 400
session.commit()
return jsonify({'batchref': batchref}), 201But our Flask app is starting to look a bit unwieldy. And our number of E2E tests is starting to get out of control, and soon we’ll end up with an inverted test pyramid (or "ice cream cone model" as Bob likes to call it).
If we look at what our Flask app is doing, there’s quite a lot of what we might call orchestration—fetching stuff out of our repository, validating our input against database state, handling errors, and committing in the happy path. Most of these things aren’t anything to do with having a web API endpoint (you’d need them if you were building a CLI for example, see [appendix_csvs]), and they’re not really things that need to be tested by end-to-end tests.
It often makes sense to split out a Service Layer, sometimes called orchestration layer or use case layer.
Do you remember the FakeRepository that we prepared in the last chapter?
class FakeRepository(repository.AbstractRepository):
def __init__(self, batches):
self._batches = set(batches)
def add(self, batch):
self._batches.add(batch)
def get(self, reference):
return next(b for b in self._batches if b.reference == reference)
def list(self):
return list(self._batches)Here’s where it will come in useful; it lets us test our service layer with nice, fast unit tests:
def test_returns_allocation():
line = model.OrderLine("o1", "COMPLICATED-LAMP", 10)
batch = model.Batch("b1", "COMPLICATED-LAMP", 100, eta=None)
repo = FakeRepository([batch]) #(1)
result = services.allocate(line, repo, FakeSession()) #(2)(3)
assert result == "b1"
def test_error_for_invalid_sku():
line = model.OrderLine("o1", "NONEXISTENTSKU", 10)
batch = model.Batch("b1", "AREALSKU", 100, eta=None)
repo = FakeRepository([batch]) #(1)
with pytest.raises(services.InvalidSku, match="Invalid sku NONEXISTENTSKU"):
services.allocate(line, repo, FakeSession()) #(2)(3)-
FakeRepositoryholds theBatchobjects that will be used by our test. -
Our services module (services.py) will define an
allocate()service-layer function. It will sit between ourallocate_endpoint()function in the API layer and theallocate()domain service function from our domain model.[1]. -
We also need a
FakeSessionto fake out the database session, see below:
class FakeSession():
committed = False
def commit(self):
self.committed = TrueThis fake session is only a temporary solution. We’ll get rid of it and make
things even nicer soon, in [chapter_06_uow]. But in the meantime
the fake .commit() lets us migrate a third test from the E2E layer:
def test_commits():
line = model.OrderLine('o1', 'OMINOUS-MIRROR', 10)
batch = model.Batch('b1', 'OMINOUS-MIRROR', 100, eta=None)
repo = FakeRepository([batch])
session = FakeSession()
services.allocate(line, repo, session)
assert session.committed is TrueWe’ll get to a service function that looks something like Basic allocation service (services.py):
class InvalidSku(Exception):
pass
def is_valid_sku(sku, batches): #(2)
return sku in {b.sku for b in batches}
def allocate(line: OrderLine, repo: AbstractRepository, session) -> str:
batches = repo.list() #(1)
if not is_valid_sku(line.sku, batches): #(2)
raise InvalidSku(f'Invalid sku {line.sku}')
batchref = model.allocate(line, batches) #(3)
session.commit() #(4)
return batchrefTypical service-layer functions have similar steps:
-
We fetch some objects from the repository
-
We make some checks or assertions about the request against the current state of the world
-
We call a domain service
-
And if all is well, we save/update any state we’ve changed.
That last step is a little unsatisfactory at the moment, our services layer is tightly coupled to our database layer, but again, we can improve that if we introduce [chapter_06_uow].
Notice one more thing about our service-layer function:
def allocate(line: OrderLine, repo: AbstractRepository, session) -> str: #(1)It depends on a repository. We’ve chosen to make the dependency explicit,
and we’ve used the type hint to say that we depend on AbstractRepository.
This means it’ll work both when the tests give it a FakeRepository, and
when the flask app gives it a SqlAlchemyRepository.
If you remember the Dependency Inversion Principle section from the introduction, this is what we mean when we says we should "depend on abstractions". Our high-level module, the service layer, depends on the repository abstraction. And the details of the implementation for our specific choice of persistent storage also depend on that same abstraction.
See the diagrams at the end of the chapter, Abstract dependencies of the service layer.
See also [appendix_csvs] where we show a worked example of swapping out the details of which persistent storage system to use, while leaving the abstractions intact.
But the essentials of the services layer are there, and our Flask app now looks a lot cleaner, Flask app delegating to service layer (flask_app.py):
@app.route("/allocate", methods=['POST'])
def allocate_endpoint():
session = get_session() #(1)
repo = repository.SqlAlchemyRepository(session) #(1)
line = model.OrderLine(
request.json['orderid'], #(2)
request.json['sku'], #(2)
request.json['qty'], #(2)
)
try:
batchref = services.allocate(line, repo, session) #(2)
except (model.OutOfStock, services.InvalidSku) as e:
return jsonify({'message': str(e)}), 400 (3)
return jsonify({'batchref': batchref}), 201 (3)We see that the responsibilities of the Flask app are much more minimal, and more focused on just the web stuff:
-
We instantiate a database session and some repository objects.
-
We extract the user’s commands from the web request and pass them to a domain service.
-
And we return some JSON responses with the appropriate status codes
The responsibilities of the Flask app are just standard web stuff: per-request session management, parsing information out of POST parameters, response status codes and JSON. All the orchestration logic is in the use case / service layer, and the domain logic stays in the domain.
Finally we can confidently strip down our E2E tests to just two, one for the happy path and one for the unhappy path:
@pytest.mark.usefixtures('restart_api')
def test_happy_path_returns_201_and_allocated_batch(add_stock):
sku, othersku = random_sku(), random_sku('other')
batch1, batch2, batch3 = random_batchref(1), random_batchref(2), random_batchref(3)
add_stock([
(batch1, sku, 100, '2011-01-02'),
(batch2, sku, 100, '2011-01-01'),
(batch3, othersku, 100, None),
])
data = {'orderid': random_orderid(), 'sku': sku, 'qty': 3}
url = config.get_api_url()
r = requests.post(f'{url}/allocate', json=data)
assert r.status_code == 201
assert r.json()['batchref'] == batch2
@pytest.mark.usefixtures('restart_api')
def test_unhappy_path_returns_400_and_error_message():
unknown_sku, orderid = random_sku(), random_orderid()
data = {'orderid': orderid, 'sku': unknown_sku, 'qty': 20}
url = config.get_api_url()
r = requests.post(f'{url}/allocate', json=data)
assert r.status_code == 400
assert r.json()['message'] == f'Invalid sku {unknown_sku}'We’ve successfully split our tests into two broad categories: tests about web stuff, which we implement end-to-end; and tests about orchestration stuff, which we can test against the service layer in memory.
We’ve now got a services allocate, why not build out a service for
deallocate? We’ve added an E2E test and a few stub service-layer tests for
you to get started here:
If that’s not enough, continue into the E2E tests and flask_app.py, and refactor the Flask adapter to be more RESTful. Notice how doing so doesn’t require any change to our service layer or domain layer!
|
Tip
|
If you decide you want to build a read-only endpoint for retrieving allocation
info, just do the simplest thing that can possibly work ™, which is
repo.get() right in the Flask handler. We’ll talk more about reads vs
writes in [chapter_12_cqrs].
|
Some of you are probably scratching your heads at this point trying to figure out exactly what is the difference between a Domain Service and a Service Layer.
We’re sorry, we didn’t choose the names, or we’d have much cooler and friendlier ways to talk about this stuff.
We’re using two things called a "service" in this chapter. The first is an Application Service (our service layer). Its job is to handle requests from the outside world, and to orchestrate an operation. What we mean is that the service layer drives the application, by following a bunch of simple steps:
-
Get some data from the database
-
Update the domain model
-
Persist any changes
This is the kind of boring work that has to happen for every operation in your system, and keeping it separate from business logic helps to keep things tidy.
The second type of service is a Domain Service. This is the name for a piece of logic that belongs in the domain model but doesn’t sit naturally inside a stateful entity or value object. For example, if you were building a shopping cart application, you might choose to build taxation rules as a Domain Service. Calculating tax is a separate job from updating the cart, and it’s an important part of the model, but it doesn’t seem to right to have a persisted entity for the job. Instead a stateless TaxCalculator class, or a calculate_tax function can do the job.
.
├── config.py
├── domain #(1)
│ ├── __init__.py
│ └── model.py
├── service_layer #(2)
│ ├── __init__.py
│ └── services.py
├── adapters #(3)
│ ├── __init__.py
│ ├── orm.py
│ └── repository.py
├── entrypoints (4)
│ ├── __init__.py
│ └── flask_app.py
└── tests
├── __init__.py
├── conftest.py
├── unit
│ ├── test_allocate.py
│ ├── test_batches.py
│ └── test_services.py
├── integration
│ ├── test_orm.py
│ └── test_repository.py
└── e2e
└── test_api.py-
Let’s have a folder for our domain model. Currently that’s just one file, but for a more complex application you might have one file per class, you might have some helper parent classes for
Entity,ValueObjectandAggregate, you might add an exceptions.py for domain-layer exceptions, and as we’ll see in [part2], commands.py and events.py. -
We’ll distinguish the service layer. Currently that’s just one file called services.py for our service-layer functions. You could add service-layer exceptions in here, and as we’ll see in the next chapter, we’ll add unit_of_work.py
-
Adapters is a nod to the Ports and Adapters terminology. This will fill up with any other abstractions around external I/O, eg a redis_client.py. Strictly speaking you would call these secondary adapters or driven adapters, or sometimes inward-facing adapters.
-
Entrypoints are the places we drive our application from. In the official Ports & Adapters terminology, these are adapters too, and are referred to as primary, driving or outward-facing adapters.
What about ports? As you may remember, they are the abstract interfaces which the adapters implement. We tend to keep them in the same file as the adapters which implement them.
Adding the service layer has really bought us quite a lot:
-
Our Flask API endpoints become very thin and easy to write: their only responsibility is doing "web stuff," things like parsing JSON and producing the right HTTP codes for happy or unhappy cases.
-
We’ve defined a clear API for our domain, a set of use cases or entrypoints that can be used by any adapter without needing to know anything about our domain model classes—whether that’s an API, a CLI (see [appendix_csvs]), or the tests! They’re an adapter for our domain too.
-
We can write tests in "high gear" using the service layer, leaving us free to refactor the domain model in any way we see fit. As long as we can still deliver the same use cases, we can experiment with new designs without needing to rewrite a load of tests.
-
And our "test pyramid" is looking good—the bulk of our tests are fast unit tests, with just the bare minimum of E2E and integration tests.
Abstract dependencies of the service layer shows the
dependencies of our service layer: the Domain Model,
and the AbstractRepository (the port, in ports & adapters terminology).
[ditaa, apwp_0403]
+-----------------------------+
| Service Layer |
+-----------------------------+
| |
| | depends on abstraction
V V
+------------------+ +--------------------+
| Domain Model | | AbstractRepository |
| | | (Port) |
+------------------+ +--------------------+
When we run the tests, Tests provide an implementation of the abstract dependency shows
how we implement the abstract dependencies using FakeRepository (the
adapter):
[ditaa, apwp_0404]
+-----------------------------+
| Tests |-------------\
+-----------------------------+ |
| |
V |
+-----------------------------+ |
| Service Layer | provides |
+-----------------------------+ |
| | |
V V |
+------------------+ +--------------------+ |
| Domain Model | | AbstractRepository | |
+------------------+ +--------------------+ |
^ |
implements | |
| |
+----------------------+ |
| FakeRepository |<--/
| (in-memory) |
+----------------------+
And when we actually run our app, we swap in the "real" dependency, Dependencies at runtime:
[ditaa, apwp_0405]
+--------------------------------+
| Flask API (Presentation layer) |-----------\
+--------------------------------+ |
| |
V |
+-----------------------------+ |
| Service Layer | |
+-----------------------------+ |
| | |
V V |
+------------------+ +--------------------+ |
| Domain Model | | AbstractRepository | |
+------------------+ +--------------------+ |
^ ^ |
| | |
gets | +----------------------+ |
model | | SqlAlchemyRepository |<--/
definitions| +----------------------+
from | | uses
| V
+-----------------------+
| ORM |
| (another abstraction) |
+-----------------------+
|
| talks to
V
+------------------------+
| Database |
+------------------------+
Wonderful. But there are still some bits of awkwardness to tidy up:
-
the service layer is still tightly coupled to the domain, because its API is expressed in terms of
OrderLineobjects. In [chapter_05_high_gear_low_gear] we’ll fix that, and talk about the way that Service Layer enables more productive TDD. -
The service layer is tightly coupled to a
sessionobject. In [chapter_06_uow] we’ll introduce one more pattern that works closely with Repository and Service Layer, the Unit of Work, and everything will be absolutely lovely. You’ll see!
But first, as is customary, a pause for Service Layer: The Tradeoffs, in which we consider the pros and cons of having a Service Layer at all.
| Pros | Cons |
|---|---|
|
|