Skip to content

The Service Layer Pattern

Overview

If we look at what the api with the repository endpoints are doing, we'll agree that are very simple. But what usually happens is that business applications typically require different kinds of interfaces (API, user interface, command line client, etc) to the data and the logic they implement (3rd party integrations, repositories,...).

The same concept used in the repository pattern to create a common interface for the data, can be used to define a common interaction with the application to invoke the business logic: fetching data from the repository, validating inputs against current state, handling errors, etc. And most of this things don't have anything to do with having a web aPI endpoint.

This is what we might call orchestration layer , use-case layer or service layer.

Example

To continue with our simple examples, we'll create a service to paginate results.

Let's start creating the test file, as usual: 

touch tests/test_04_api_with_services_and_repository.py

First Test and Dependency Injection

To implement the service layer, we know that we'll need a mechanism to inject the in-memory repository to the service. To do this, we define a function get_repo_test that returns an InMemoryBookRepository with a list of 25 new books, that will be used to return paginated results.

The test, overrides the app dependencies with this repository, and do a request to a new endpoint, asking for a page of results.

tests/test_04_api_with_services_and_repository.py

@@ -0,0 +1,32 @@

+import pytest
+
+from tests.repositories import InMemoryBookRepository
+from tests.schemas import Book
+
+
+def get_repo_test() -> InMemoryBookRepository:
+    repo_test = InMemoryBookRepository(initial_books=dict())
+    for book_id in range(0, 25):
+        repo_test.add(Book(title=f"title {book_id}", author=f"author {book_id}"))
+    return repo_test
+
+
+@pytest.mark.parametrize(
+    "page_num, results_count, expected_status", (
+        (1, 10, 200),
+        (3, 5, 200),
+        (5, 0, 200),
+        (1000, 0, 200),
+    ),
+)
+def test_pagination_service_successfully(app, client, page_num, results_count, expected_status) -> None:
+    app.dependency_overrides[InMemoryBookRepository] = get_repo_test
+
+    response = client.get(f"/books?page_num={page_num}")
+    assert response.status_code == expected_status
+
+    content = response.json()
+    assert "page" in content
+    assert content["page"] == page_num
+    assert "content" in content
+    assert len(content["content"]) == results_count

Of course, this test fails, because the endpoint is not yet defined.

Create Service and Endpoint

At this point, we can imagine that the new service will be used from other tests, and perhaps in the future, from a real application. for this reason, we create it in a new file:

{git_diff_by_commit a5f556cd..8f447f5 - tests / services.py}

Since we have a conftest.py with an app fixture, we'll continue to extend it with the new endpoint, which uses the previous service.

tests/conftest.py

@@ -1,19 +1,20 @@

 from typing import List
 from uuid import UUID

 import pytest
 from fastapi import FastAPI, Depends
 from starlette.testclient import TestClient

 from tests.repositories import InMemoryBookRepository
 from tests.schemas import Book
+from tests.services import BookPaginationService


 @pytest.fixture
 def app():
     app = FastAPI()

     @app.get("/", name="books:get-all-books")
     def get_all_books(
         books_repo: InMemoryBookRepository = Depends(InMemoryBookRepository)
     ) -> List[Book]:
@@ -29,20 +30,28 @@

         return book

     @app.post("/", name="books:create-book", status_code=201)
     def create_new_book(
         book: Book,
         books_repo: InMemoryBookRepository = Depends(InMemoryBookRepository)
     ) -> Book:
         book_db = books_repo.add(book)
         return book_db

+    @app.get("/books")
+    def get_paginated_books(
+        page_num: int = 1,
+        book_pagination: BookPaginationService = Depends(BookPaginationService),
+    ):
+        paginated_books = book_pagination.get_page(num=page_num)
+        return paginated_books
+
     return app


 @pytest.fixture
 def client(app):
     return TestClient(app)


 @pytest.fixture()
 def a_book() -> Book:

Update Repository

We need that our repository returns only the requested items. To do this, we modify the list method to accept pagination params:

tests/repositories.py

@@ -30,12 +30,12 @@

     def add(self, new_book: Book) -> BookInDB:
         if not isinstance(new_book, Book):
             raise ValueError("This repository only accepts Book objects.")
         book = BookInDB(**new_book.dict())
         self.books[book.id] = book
         return book

     def get(self, book_id: UUID) -> Optional[BookInDB]:
         return self.books.get(book_id, None)

-    def list(self) -> List[BookInDB]:
-        return [book for book_id, book in self.books.items()]
+    def list(self, skip=0, offset=5) -> List[BookInDB]:
+        return list(self.books.values())[skip * offset:(skip + 1) * offset]

Add Test for Invalid Values

We need to be sure that the system is as robust as possible, and that handle correctly invalid values for pagination requests. Let's add a test to check invalid values, using the parametrized decorator:

tests/test_04_api_with_services_and_repository.py

@@ -23,10 +23,25 @@

     app.dependency_overrides[InMemoryBookRepository] = get_repo_test

     response = client.get(f"/books?page_num={page_num}")
     assert response.status_code == expected_status

     content = response.json()
     assert "page" in content
     assert content["page"] == page_num
     assert "content" in content
     assert len(content["content"]) == results_count
+
+
+@pytest.mark.parametrize(
+    "page_num, expected_status", (
+        (-1, 422),
+        (0, 422),
+        ('0', 422),
+        ('a', 422),
+    ),
+)
+def test_pagination_service_fails(app, client, page_num, expected_status) -> None:
+    app.dependency_overrides[InMemoryBookRepository] = get_repo_test
+
+    response = client.get(f"/books?page_num={page_num}")
+    assert response.status_code == expected_status

Fix Code

With the previous test, we realize that some invalid input values are accepted. Let's fix it adding input data validation to the endpoint:

tests/conftest.py

@@ -1,15 +1,16 @@

 from typing import List
 from uuid import UUID

 import pytest
 from fastapi import FastAPI, Depends
+from pydantic.types import PositiveInt
 from starlette.testclient import TestClient

 from tests.repositories import InMemoryBookRepository
 from tests.schemas import Book
 from tests.services import BookPaginationService


 @pytest.fixture
 def app():
     app = FastAPI()
@@ -32,21 +33,21 @@

     @app.post("/", name="books:create-book", status_code=201)
     def create_new_book(
         book: Book,
         books_repo: InMemoryBookRepository = Depends(InMemoryBookRepository)
     ) -> Book:
         book_db = books_repo.add(book)
         return book_db

     @app.get("/books")
     def get_paginated_books(
-        page_num: int = 1,
+        page_num: PositiveInt = 1,
         book_pagination: BookPaginationService = Depends(BookPaginationService),
     ):
         paginated_books = book_pagination.get_page(num=page_num)
         return paginated_books

     return app


 @pytest.fixture
 def client(app):

All green!