Blogcraft
Tutorial

Clean Code Python: The Service Layer — Where Business Logic Lives

Repositories fetch data. API routes handle HTTP. Business logic belongs in neither. The Service Layer is the missing middle that keeps both clean and testable.

Tin Dang avatar
Tin Dang 7 min read
Layers of transparent glass panels stacked cleanly, representing architectural separation of concerns

When a discount calculation ends up in a route handler, you have already made a mistake you will pay for at 2am when you need to reuse that logic somewhere else. The route handler becomes a grab-bag of business rules, validation, HTTP concerns, and database calls all tangled together. Adding the same discount logic to a second endpoint means copy-paste — and a future bug fix that misses one of the two locations.

Services are the antidote. A service class holds all business logic for a domain entity. Route handlers call services. Services call repositories. The business rule lives in exactly one place.

What Goes in a Service (and What Doesn’t)

The hardest part of the Service Layer pattern is enforcing the boundaries. Most business logic is obvious — it goes in the service. But HTTP status codes? Validation? Database sessions? Those have homes elsewhere.

Belongs in ServiceBelongs in RepositoryBelongs in Route
Business rules (discounts, stock checks) Data fetching by ID or criteria HTTP status code selection
Input domain validation (ISBN format, price range) Insert, update, delete operations Request/response serialization
Transaction orchestration across repos Query optimization (selectinload, indexes) Authentication dependency injection
Domain exception raising (BookNotFoundError) IntegrityError → domain exception mapping Domain exception → HTTPException translation
Notification triggers (email, webhook) Raw SQL / ORM session management Response model construction

The pattern: services raise domain exceptions, routes translate them to HTTP responses. A service that raises HTTPException has leaked an HTTP concern into the business layer.

The Architecture: Where Services Fit

The Place Order flow illustrates how every layer plays its role without overlapping another’s responsibilities:

The route never touches business logic. The service never constructs an HTTP response. The repositories never know about discounts.

Implementing BookService

Here is a complete BookService with real business rules: bulk discounts, ISBN validation, and domain exception raising.

src/services/book_service.py
from decimal import Decimal
from src.repositories.protocols import BookRepositoryProtocol
from src.schemas.book import BookCreate, BookResponse
from src.core.exceptions import BookNotFoundError, InvalidISBNError


BULK_DISCOUNT_THRESHOLD = 10
BULK_DISCOUNT_RATE = Decimal("0.15")




class BookService:
    def __init__(self, repo: BookRepositoryProtocol) -> None:
        self._repo = repo


    async def get_book(self, book_id: int) -> BookResponse:
        book = await self._repo.get_by_id(book_id)
        if book is None:
            raise BookNotFoundError(book_id)
        return book


    async def add_book(self, data: BookCreate) -> BookResponse:
        if not self._validate_isbn(data.isbn):
            raise InvalidISBNError(data.isbn)
        return await self._repo.create(data)


    async def calculate_order_total(self, book_ids: list[int]) -> Decimal:
        books = [await self._repo.get_by_id(bid) for bid in book_ids]
        valid_books = [b for b in books if b is not None]
        total = sum(b.price for b in valid_books)
        if len(book_ids) >= BULK_DISCOUNT_THRESHOLD:
            total = total * (1 - BULK_DISCOUNT_RATE)
        return total


    def _validate_isbn(self, isbn: str) -> bool:
        digits = isbn.replace("-", "")
        return len(digits) in (10, 13) and digits.isdigit()

The _validate_isbn method is a private helper — ISBN format is a domain rule (a business concern), not a schema validation concern. Pydantic validates that isbn is a string; the service validates that the string is a real ISBN.

Custom Domain Exceptions

Services raise domain exceptions. Routes catch them and translate to HTTP responses. For this handoff to work cleanly, every domain error needs its own exception class with typed fields.

src/core/exceptions.py
class BookStoreError(Exception):
    """Base exception for all domain errors."""




class BookNotFoundError(BookStoreError):
    def __init__(self, book_id: int) -> None:
        self.book_id = book_id
        super().__init__(f"Book {book_id} not found")




class InvalidISBNError(BookStoreError):
    def __init__(self, isbn: str) -> None:
        self.isbn = isbn
        super().__init__(f"Invalid ISBN: {isbn}")




class InsufficientStockError(BookStoreError):
    def __init__(self, book_id: int, available: int) -> None:
        self.book_id = book_id
        self.available = available
        super().__init__(f"Book {book_id} has only {available} copies")




class DuplicateISBNError(BookStoreError):
    def __init__(self, isbn: str) -> None:
        self.isbn = isbn
        super().__init__(f"ISBN {isbn} already exists")

The base BookStoreError lets route handlers catch any domain error with a single except BookStoreError when they only need to return a generic 400. Specific handlers can still match BookNotFoundError for a 404. Python’s exception hierarchy gives you precision without redundancy.

OrderService: Multi-Repository Orchestration

The most valuable service methods orchestrate multiple repositories in a single transaction. OrderService.place_order() touches two repositories while guaranteeing atomicity: either the stock decrement and the order creation both succeed, or neither does.

src/services/order_service.py
from decimal import Decimal
from sqlalchemy.ext.asyncio import AsyncSession
from src.repositories.protocols import OrderRepositoryProtocol, BookRepositoryProtocol
from src.schemas.order import OrderResponse
from src.core.exceptions import BookNotFoundError, InsufficientStockError


BULK_DISCOUNT_THRESHOLD = 10
BULK_DISCOUNT_RATE = Decimal("0.15")




class OrderService:
    def __init__(
        self,
        order_repo: OrderRepositoryProtocol,
        book_repo: BookRepositoryProtocol,
        session: AsyncSession,
    ) -> None:
        self._order_repo = order_repo
        self._book_repo = book_repo
        self._session = session


    async def place_order(self, user_id: int, book_ids: list[int]) -> OrderResponse:
        async with self._session.begin():
            # Validate all books exist and have stock
            for book_id in book_ids:
                book = await self._book_repo.get_by_id(book_id)
                if book is None:
                    raise BookNotFoundError(book_id)
                if book.stock < 1:
                    raise InsufficientStockError(book_id, book.stock)


            total = await self._calculate_total(book_ids)
            order = await self._order_repo.create(
                user_id=user_id,
                book_ids=book_ids,
                total=total,
            )


        return order


    async def _calculate_total(self, book_ids: list[int]) -> Decimal:
        books = [await self._book_repo.get_by_id(bid) for bid in book_ids]
        valid = [b for b in books if b is not None]
        total = sum(b.price for b in valid)
        if len(book_ids) >= BULK_DISCOUNT_THRESHOLD:
            total = total * (1 - BULK_DISCOUNT_RATE)
        return total

The transaction boundary lives in the service, not in the repository. Each repository method participates in an existing transaction by receiving a session that already has an active begin(). When the service raises BookNotFoundError inside the async with self._session.begin() block, SQLAlchemy automatically rolls back — the stock decrement that may have already happened in the same transaction is undone.

Testing the Service in Isolation

With BookRepositoryProtocol as the service’s dependency, tests substitute FakeBookRepository and never touch a database.

tests/test_book_service.py
import pytest
from decimal import Decimal
from src.services.book_service import BookService
from src.core.exceptions import BookNotFoundError, InvalidISBNError
from tests.fakes import FakeBookRepository




@pytest.fixture
def service() -> BookService:
    return BookService(repo=FakeBookRepository())




@pytest.mark.asyncio
async def test_get_book_raises_when_not_found(service: BookService) -> None:
    with pytest.raises(BookNotFoundError) as exc_info:
        await service.get_book(book_id=999)
    assert exc_info.value.book_id == 999




@pytest.mark.asyncio
async def test_add_book_rejects_invalid_isbn(service: BookService) -> None:
    from src.schemas.book import BookCreate
    with pytest.raises(InvalidISBNError):
        await service.add_book(BookCreate(title="Test", isbn="not-an-isbn", price=Decimal("9.99")))




@pytest.mark.asyncio
async def test_bulk_discount_applied(service: BookService) -> None:
    # Seed 11 books into the fake repository
    repo = FakeBookRepository(seed_count=11, unit_price=Decimal("10.00"))
    bulk_service = BookService(repo=repo)
    book_ids = list(range(1, 12))
    total = await bulk_service.calculate_order_total(book_ids)
    # 11 * 10.00 = 110.00, with 15% discount = 93.50
    assert total == Decimal("93.50")




@pytest.mark.asyncio
async def test_no_discount_below_threshold(service: BookService) -> None:
    repo = FakeBookRepository(seed_count=5, unit_price=Decimal("10.00"))
    svc = BookService(repo=repo)
    total = await svc.calculate_order_total([1, 2, 3])
    assert total == Decimal("30.00")

These tests run in milliseconds. No test database, no test container, no migration. The service’s contract with BookRepositoryProtocol is the only coupling point, and FakeBookRepository satisfies it completely.

Key Takeaways

  • Services own business logic — discounts, validation rules, stock checks, notifications. Business logic that lives in a route handler is untestable and unreusable.
  • Raise domain exceptions, not HTTP errors — BookNotFoundError, not HTTPException(status_code=404). Services are framework-agnostic.
  • Accept Protocols, not concrete implementations — BookRepositoryProtocol over BookRepository makes every service method independently testable with an in-memory fake.
  • Orchestrate transactions across repositories — the service owns the session.begin() boundary; repositories participate without knowing about each other.
  • Zero-dependency tests — service tests are unit tests. No HTTP client, no database, no fixtures beyond the fake repository.

Next: Part 5 shows how FastAPI’s Depends() wires repositories and services into route handlers without global state or service locators.

0

Next in this series

Clean Code Python: Dependency Injection Without Magic

Continue reading

Related Posts