25.5. Service Layer
Table of Contents
Why a Service Layer Exists
In most backend applications you do not want your routes or controllers to talk directly to the database or external services. That makes the code hard to change, test, and understand.
A service layer is a set of classes or functions that contain your application’s business logic. Routes call services, and services talk to repositories, external APIs, queues, and other infrastructure.
A very simple mental picture:
[ HTTP Route / Controller ]
↓
[ Service Layer ]
↓
[ Repositories, APIs, etc. ]
↓
[ Database ]The service layer answers questions like:
- “What happens when a user registers?”
- “How do we create an order?”
- “What checks happen when someone cancels an order?”
It is not concerned with HTTP details, response models, or SQL syntax. It is concerned with rules and workflows.
Key rule: Put business rules and application workflows in the service layer, not in controllers or repositories.
Responsibilities of the Service Layer
The service layer sits between the web / API layer and the data access layer. It has some very specific responsibilities.
Orchestrating Use Cases
Each service method usually corresponds to a use case or application action. For example:
register_userlogin_usercreate_ordercancel_orderadd_item_to_cart
Each of these methods coordinates multiple steps:
Example: create_order
- Load the user and cart from the database.
- Check inventory for each product.
- Calculate prices, taxes, and discounts.
- Create an order record.
- Reserve inventory.
- Publish an “order_created” event to a message queue.
- Return the created order.
In a service-oriented design, the controller does something like:
order = order_service.create_order(user_id=user.id)
return OrderResponse.from_order(order)The controller does not know how to create the order. It just asks the service.
Enforcing Business Rules
Business rules describe how your system should behave. Examples:
- “A user cannot place an order if their account is banned.”
- “An order cannot be shipped if payment is not captured.”
- “A password must be at least 12 characters long.”
These rules live in the service layer (or in domain entities, if you use DDD), not in the controller or repository.
Example in pseudo Python:
class OrderService:
def ship_order(self, order_id: int, current_user: User):
order = self.order_repo.get(order_id)
if not current_user.is_admin:
raise PermissionError("Only admins can ship orders")
if order.status != "paid":
raise ValueError("Order must be paid before shipping")
order.status = "shipped"
self.order_repo.save(order)
self.events.publish("order_shipped", order_id=order.id)The controller might only handle:
try:
order_service.ship_order(order_id, current_user)
return {"status": "ok"}
except PermissionError:
raise HTTPException(status_code=403)
except ValueError as exc:
raise HTTPException(status_code=400, detail=str(exc))Rules live in the service, HTTP mapping lives in the controller.
Coordinating Multiple Repositories and Services
Often a use case touches multiple tables or external systems. The service layer is the place where you coordinate them.
Example: “User deletes their account”
Steps:
- Mark user as deleted.
- Cancel active subscriptions via payment gateway.
- Remove user sessions from Redis.
- Anonymize or delete user data in analytics.
- Send confirmation email.
A service method might look like:
class AccountService:
def delete_account(self, user_id: int):
user = self.user_repo.get(user_id)
self.payment_gateway.cancel_subscriptions(user)
self.session_store.remove_sessions(user_id)
self.analytics_service.anonymize_user(user_id)
user.is_deleted = True
self.user_repo.save(user)
self.email_service.send_account_deleted_email(user.email)No HTTP, no SQL queries here, only coordination of other abstractions.
Managing Transactions at the Use Case Level
The service layer is the right place to decide when a transaction begins and ends.
Example:
class OrderService:
def __init__(self, order_repo, inventory_repo, unit_of_work):
self.order_repo = order_repo
self.inventory_repo = inventory_repo
self.uow = unit_of_work
def create_order(self, user_id: int) -> Order:
with self.uow.transaction():
order = self.order_repo.create_empty(user_id)
items = self.inventory_repo.get_cart_items(user_id)
for item in items:
if item.stock <= 0:
raise ValueError("Out of stock")
self.inventory_repo.reserve(item)
order.items = items
self.order_repo.save(order)
return order
Inside with self.uow.transaction(): all repository calls share the same database transaction.
Important: Transactions should usually wrap a whole use case, not individual repository calls, so that either the entire operation succeeds or it all fails.
What the Service Layer Should Not Do
To keep boundaries clear, it helps to know what does not belong in the service layer.
No HTTP Details
Avoid:
RequestorResponseobjects.- Query parameters and path parameters.
- HTTP status codes.
A bad example:
# Bad: service returning HTTP exceptions
class UserService:
def register(self, request: Request):
if "email" not in request.json:
raise HTTPException(400, "Missing email")Better:
# Good: service deals with plain data and domain concepts
class UserService:
def register(self, email: str, password: str) -> User:
if not email:
raise ValueError("Email is required")
# ...
return userController:
def register_user_endpoint(body: RegisterRequest):
try:
user = user_service.register(body.email, body.password)
except ValueError as exc:
raise HTTPException(status_code=400, detail=str(exc))
return UserResponse.from_user(user)No SQL or ORM Queries
The service layer should talk to repositories or data access objects, not to the ORM or database directly.
Bad:
class OrderService:
def list_orders(self):
return session.query(Order).filter(Order.status == "paid").all()Good:
class OrderService:
def list_paid_orders(self):
return self.order_repo.list_by_status("paid")Repository:
class SqlAlchemyOrderRepository:
def list_by_status(self, status: str) -> list[Order]:
return (
self.session.query(Order)
.filter(Order.status == status)
.all()
)This makes it easier to swap databases or to test the service with in-memory repositories.
No Framework-Specific Logic
Try to keep the service layer independent of your web framework. No FastAPI decorators, no Flask globals, etc.
This helps when:
- You change frameworks later.
- You want to reuse logic in a CLI tool or batch script.
- You write unit tests without spinning up the web server.
Example: Service Layer in a Simple Application
Imagine a very small task management API. You might have layers like this:
| Layer | Example components | Knows about |
|---|---|---|
| Presentation | FastAPI routes, request & response models | HTTP, JSON, auth tokens |
| Service layer | TaskService, UserService | Business rules, use cases |
| Data access layer | Repositories, ORM models | Database, SQL / ORM |
| Infrastructure | Email sender, payment gateway client, Redis | External systems, protocols |
A simple workflow: “Create a task”
Controller (FastAPI endpoint)
@router.post("/tasks", response_model=TaskResponse)
def create_task(
body: CreateTaskRequest,
current_user: User = Depends(get_current_user),
):
task = task_service.create_task(
title=body.title,
description=body.description,
owner_id=current_user.id,
)
return TaskResponse.from_task(task)Service layer
class TaskService:
def __init__(self, task_repo, event_bus):
self.task_repo = task_repo
self.event_bus = event_bus
def create_task(self, title: str, description: str, owner_id: int) -> Task:
if not title or len(title) < 3:
raise ValueError("Title must be at least 3 characters")
task = Task(title=title, description=description, owner_id=owner_id)
self.task_repo.add(task)
self.event_bus.publish("task_created", {"task_id": task.id})
return taskRepository
class SqlAlchemyTaskRepository:
def __init__(self, session):
self.session = session
def add(self, task: Task):
self.session.add(task)
self.session.commit()Notice:
- Controllers handle HTTP concerns.
- Service handles validation and orchestration.
- Repository handles persistence.
Designing Service Interfaces
How you design the methods in your service layer affects readability and usability.
Use Use-Case Oriented Methods
Service methods should sound like actions or use cases, not low-level operations.
Compare:
- Bad:
get,save,update,delete - Better:
register_user,change_password,add_product_to_cart,checkout_cart
Good names make controllers satisfy a rule like: “One endpoint calls one service method.”
Example mapping:
| Endpoint | Service method |
|---|---|
POST /users | user_service.register_user |
POST /login | auth_service.login |
POST /orders | order_service.create_order |
PATCH /orders/{id} | order_service.update_order |
POST /orders/{id}/pay | payment_service.pay_order |
Accept and Return Domain Types
Prefer passing simple values and domain objects, not framework objects.
Bad:
def create_order(self, request: Request) -> Response:
# service is tied to HTTPGood:
def create_order(
self,
user_id: int,
items: list[OrderItemInput],
) -> Order:
# no HTTP knowledgeIf you use data classes or Pydantic models for domain data, create separate request / response models in the API layer, and convert between them.
Handle Errors with Exceptions
Let the service layer raise domain-specific exceptions and let the controller map them to HTTP responses.
Example:
class OutOfStockError(Exception):
pass
class OrderNotFoundError(Exception):
pass
class OrderService:
def cancel_order(self, order_id: int, user_id: int):
order = self.order_repo.get(order_id)
if not order:
raise OrderNotFoundError()
if order.user_id != user_id:
raise PermissionError("Not your order")
if order.status == "shipped":
raise ValueError("Cannot cancel shipped order")
order.status = "cancelled"
self.order_repo.save(order)Controller:
@router.post("/orders/{order_id}/cancel")
def cancel_order(order_id: int, current_user: User = Depends(get_current_user)):
try:
order_service.cancel_order(order_id, current_user.id)
except OrderNotFoundError:
raise HTTPException(status_code=404, detail="Order not found")
except PermissionError:
raise HTTPException(status_code=403, detail="Forbidden")
except ValueError as exc:
raise HTTPException(status_code=400, detail=str(exc))
return {"status": "cancelled"}Service Layer and Other Architectural Patterns
The service layer often appears together with other patterns that you have in the outline.
With the Repository Pattern
The repository pattern abstracts persistence. The service layer uses repositories instead of using ORM sessions or SQL directly.
Typical dependencies:
class UserService:
def __init__(self, user_repo: UserRepository, email_service: EmailService):
self.user_repo = user_repo
self.email_service = email_serviceRepositories hide the details of PostgreSQL or any other database. Services focus on rules like “user must verify email before logging in.”
With Domain-Driven Design (DDD)
In DDD you will see terms like:
- Entities: domain objects with identity.
- Value objects: small immutable value types.
- Domain services: operations that do not naturally belong to an entity.
You can think of the “service layer” in two ways:
- Application services: orchestrate use cases, call repositories, manage transactions.
- Domain services: pure business logic operations that work with entities and value objects.
A typical setup:
# Application service
class PaymentApplicationService:
def pay_order(self, order_id: int, payment_details: PaymentDetails):
order = self.order_repo.get(order_id)
payment = self.payment_domain_service.pay(order, payment_details)
self.payment_repo.save(payment)The important point is that the application service is still the boundary between the outside world and the domain.
With Dependency Injection
Services are often wired with dependency injection:
- In small apps, you may manually instantiate services in a module and reuse them.
- In larger apps, you may use a DI container.
Example with manual wiring in a FastAPI app:
# wiring.py
session_factory = create_session_factory()
user_repo = SqlAlchemyUserRepository(session_factory)
email_service = SmtpEmailService(settings.smtp)
user_service = UserService(user_repo=user_repo, email_service=email_service)Route:
@router.post("/users")
def register_user(body: RegisterRequest):
user = user_service.register(body.email, body.password)
return UserResponse.from_user(user)DI keeps your service constructors explicit, which makes them easy to test.
Testing the Service Layer
The service layer is a sweet spot for unit tests, because it:
- Contains important logic.
- Can often be tested without HTTP or a real database.
Testing with In-Memory or Fake Repositories
Instead of using the real database, you can write a simple in-memory repository that implements the same interface.
Example:
class InMemoryUserRepository:
def __init__(self):
self._users = {}
self._next_id = 1
def add(self, user: User):
user.id = self._next_id
self._next_id += 1
self._users[user.id] = user
def get_by_email(self, email: str) -> User | None:
return next((u for u in self._users.values() if u.email == email), None)Test:
def test_register_user_creates_user_and_sends_email():
repo = InMemoryUserRepository()
email_service = FakeEmailService()
service = UserService(user_repo=repo, email_service=email_service)
user = service.register("a@example.com", "password123")
assert user.id is not None
assert email_service.sent_to == ["a@example.com"]No HTTP request, no database, only logic being tested.
Testing Business Rules
You can write many small tests that verify the rules inside your services.
Examples of test scenarios for an order service:
- Cannot cancel shipped order.
- Only admin can change order price.
- Cannot create an order for banned user.
- Discount is applied correctly when there are more than 3 items.
These tests help you change implementation later without breaking behavior.
Rule for testable design: If it is hard to test a piece of logic without HTTP and a real database, that logic is probably not in the service layer where it belongs.
When a Service Layer Is Overkill
For very small scripts or simple prototypes, a formal service layer can feel like too much structure:
- A single-file FastAPI app that does simple CRUD.
- A temporary internal tool.
In such cases, you may allow some logic in controllers.
However, as soon as:
- You add real business rules.
- Multiple controllers share logic.
- You need solid tests.
it becomes beneficial to introduce services.
A minimal first step is often:
- Extract non-trivial logic from endpoints into plain functions like
def create_order(...). - Later, group those functions into classes like
OrderService.
Practical Guidelines
To use a service layer effectively, keep these guidelines in mind.
What to Put in the Service Layer
- Business rules.
- Input validation that is about “business correctness,” not just “is this an integer.”
- Coordination of multiple repositories or systems.
- Transaction boundaries.
- Emitting domain events or integration events.
- Permission checks that depend on business context.
What to Keep Out of the Service Layer
- HTTP objects, status codes, or headers.
- Framework-specific abstractions.
- SQL queries, ORM session usage, or raw database connections.
- HTML templates or serialization logic.
Simple Checklist
When you write a new endpoint, ask:
- Does this endpoint do more than simply call a repository?
- Are there business rules or workflows involved?
- Might this logic be reused elsewhere?
If the answer is “yes” to any of these, create or extend a service and put the logic there. Let the controller be thin and focused on HTTP concerns only.
Views: 7
KAHIBARO