Back to our allocations project! Figure 4-1 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.
Figure 4-2 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
.
In our diagrams, we are using the convention that the yellow/orange blocks mean new things that we are adding to our architecture.
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 allocate
domain 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.
Example 4-1 shows a first cut:
A first API test (test_api.py)
@pytest.mark.usefixtures
(
'
restart_api
'
)
def
test_api_returns_allocation
(
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
random_sku()
, random_batchref()
etc are little helper functions that
add generate some randomized characters using the uuid
module. 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_stock
is 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 B.
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 B.
Implementing things in the most obvious way, you might get something like this:
First cut Flask app (flask_app.py)
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
}),
201
So 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:
Test allocations are persisted (test_api.py)
@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'
]
==
batch2
Not 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:
Yet more tests at the e2e layer… (test_api.py)
@pytest.mark.usefixtures
(
'
restart_api
'
)
def
test_400_message_for_out_of_stock
(
add_stock
)
:
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
(
)
:
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:
Flask app starting to get crufty (flask_app.py)
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
}),
201
But 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 C), 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?
Our fake repository, an in-memory collection of Batches (test_services.py)
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:
Unit testing with fakes at the services layer (test_services.py)
def
test_returns_allocation
(
)
:
line
=
model
.
OrderLine
(
"
o1
"
,
"
COMPLICATED-LAMP
"
,
10
)
batch
=
model
.
Batch
(
"
b1
"
,
"
COMPLICATED-LAMP
"
,
100
,
eta
=
None
)
repo
=
FakeRepository
(
[
batch
]
)
result
=
services
.
allocate
(
line
,
repo
,
FakeSession
(
)
)
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
]
)
with
pytest
.
raises
(
services
.
InvalidSku
,
match
=
"
Invalid sku NONEXISTENTSKU
"
)
:
services
.
allocate
(
line
,
repo
,
FakeSession
(
)
)
FakeRepository
holds the Batch
objects that will be used by our test.
Our services module (services.py) will define an allocate()
service-layer function. It will sit between our allocate_endpoint()
function in the API layer and the allocate()
domain service function from
our domain model.1.
We also need a FakeSession
to fake out the database session, see below:
A fake database session (test_services.py)
class
FakeSession
():
committed
=
False
def
commit
(
self
):
self
.
committed
=
True
This fake session is only a temporary solution. We’ll get rid of it and make
things even nicer soon, in Chapter 6. But in the meantime
the fake .commit()
lets us migrate a third test from the E2E layer:
A second test at the service layer (test_services.py)
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
True
We’ll get to a service function that looks something like Example 4-10:
Basic allocation service (services.py)
class
InvalidSku
(
Exception
)
:
pass
def
is_valid_sku
(
sku
,
batches
)
:
return
sku
in
{
b
.
sku
for
b
in
batches
}
def
allocate
(
line
:
OrderLine
,
repo
:
AbstractRepository
,
session
)
-
>
str
:
batches
=
repo
.
list
(
)
if
not
is_valid_sku
(
line
.
sku
,
batches
)
:
raise
InvalidSku
(
f
'
Invalid sku {line.sku}
'
)
batchref
=
model
.
allocate
(
line
,
batches
)
session
.
commit
(
)
return
batchref
Typical 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 6.
But the essentials of the services layer are there, and our Flask app now looks a lot cleaner, Example 4-12:
Flask app delegating to service layer (flask_app.py)
@app.route
(
"
/allocate
"
,
methods
=
[
'
POST
'
]
)
def
allocate_endpoint
(
)
:
session
=
get_session
(
)
repo
=
repository
.
SqlAlchemyRepository
(
session
)
line
=
model
.
OrderLine
(
request
.
json
[
'
orderid
'
]
,
request
.
json
[
'
sku
'
]
,
request
.
json
[
'
qty
'
]
,
)
try
:
batchref
=
services
.
allocate
(
line
,
repo
,
session
)
except
(
model
.
OutOfStock
,
services
.
InvalidSku
)
as
e
:
return
jsonify
(
{
'
message
'
:
str
(
e
)
}
)
,
400
return
jsonify
(
{
'
batchref
'
:
batchref
}
)
,
201
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:
E2E tests now only happy + unhappy paths (test_api.py)
@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.
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.
Some subfolders
. ├── config.py ├── domain
│ ├── __init__.py │ └── model.py ├── service_layer
│ ├── __init__.py │ └── services.py ├── adapters
│ ├── __init__.py │ ├── orm.py │ └── repository.py ├── entrypoints
│ ├── __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
, ValueObject
and
Aggregate
, you might add an exceptions.py for domain-layer exceptions,
and as we’ll see in Part II, 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. If you really wanted to go to town, you could split out the abstract repository (which is the Port) into its own file, along with any other ABCs for other adapters, and put them in a folder called ports3.
Some subfolders
. ├── domain │ ├── __init__.py │ └── model.py ├── adapters │ ├── __init__.py │ ├── orm.py │ └── sqlalchemy_repository.py ├── ports │ ├── abstract_repository.py ...
The primary/upstream ports are harder to define at this stage, because the port is just “the service-layer function and its arguments”. In later chapter, we build Command and Event classes, and you can call those ports in theory, but putting them into a ports subfolder instead of inside the domain folder would feel weird to us.
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 C), 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.
Figure 4-3 shows the
dependencies of our service layer: the Domain Model,
and the AbstractRepository
(the port, in ports & adapters terminology).
[ditaa, service_layer_diagram_abstract_dependencies] +-----------------------------+ | Service Layer | +-----------------------------+ | | | | depends on abstraction V V +------------------+ +--------------------+ | Domain Model | | AbstractRepository | | | | (Port) | +------------------+ +--------------------+
When we run the tests, Figure 4-4 shows
how we implement the abstract dependencies using FakeRepository
(the
adapter):
[ditaa, service_layer_diagram_test_dependencies] +-----------------------------+ | 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, Figure 4-5:
[ditaa, service_layer_diagram_runtime_dependencies] +--------------------------------+ | 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 OrderLine
objects. In
Chapter 5 we’ll fix that, and talk about
the way that Service Layer enables more productive TDD.
The service layer is tightly coupled to a session
object. In Chapter 6
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 Table 4-1, in which we consider the pros and cons of having a Service Layer at all.
Pros | Cons |
---|---|
|
|
1 Service-layer services and domain services do have confusingly similar names. We have a sidebar on it later in the chapter “Why Is Everything Called A Service?”.
2 Is this Pythonic? Depending on who you ask, both abstract base classes and type hints are hideous abominations, and serve only to add useless, unreadable cruft to your code; beloved only by people who wish that Python was Haskell, which it will never be. “beautiful is better than ugly,” “simple is better than complex,” and “readability counts…” Or, perhaps they make explicit something that would otherwise be implicit (“explicit is better than implicit”). For the purposes of this book, we’ve decided this argument carries the day. What you decide to do in your own codebase is up to you.
3 Our tech reviewer David Seddon would particularly enjoy that, because it would make our module dependency graph really neat, illustrating the “depend on abstractions” concept.
18.221.239.148