Services

Services are where business logic lives.

The service layer speaks the specific domain language of the software, can access the database & other resources & can interact with other parts of your system.

Here's a very simple diagram, positioning the service layer in our apps:

Service layer A service can be: - A simple function. - A class. - An entire module. - Whatever makes sense in your case.

In most cases, a service can be simple function that:

Lives in <your_app>/services.py module.
Takes keyword-only arguments, unless it requires no or one argument.
Is type-annotated.
Interacts with the database, other resources & other parts of your system.
Does business logic -- from simple model creation to complex cross-cutting concerns, to calling external services & tasks.

An example service that creates a user:

from sqlalchemy.orm import Session

def user_create(email: str, name: str, db_session: Session) -> User:
    with db_session.begin() as session:
      user = User(email=email)
      session.add(user)

    profile_create(user=user, name=name)
    confirmation_email_send(user=user)

    return user

As you can see, this service calls two other services -- profile_create and confirmation_email_send.

In this example, everything related to the user creation is in one place and can be traced.

Naming convention

If we take the example above, our service is named user_create. The pattern is <entity>_<action>.

This seems odd at first, but it has few nice features:

Namespacing. It's easy to spot all services starting with user_ and it's a good idea to put them in a users.py module.
Greppability. Or in other words, if you want to see all actions for a specific entity, just grep for user_.

Modules

If you have a simple-enough app with a bunch of services, they can all live happily in the service.py module.

But when things get big, you might want to split services.py into a folder with sub-modules, depending on the different sub-domains that you are dealing with in your app.

For example, lets say we have an authentication app, where we have one sub-module in our services module, that deals with jwt, and one sub-module that deals with oauth.

The structure may look like this:

services
├── __init__.py
├── jwt.py
└── oauth.py

There are lots of flavors here:

You can do the import-export dance in services/__init__.py, so you can import from project.authentication.services everywhere else.
You can create a folder-module, jwt/__init__.py, and put the code there.
Basically, the structure is up to you. If you feel it's time to restructure and refactor - do so.

Selectors

In most projects, we distinguish between "Pushing data to the database" and "Pulling data from the database":

Services take care of the push.
Selectors take care of the pull.
Selectors can be viewed as a "sub-layer" to services, that's specialized in fetching data.

A selector follows the same rules as a service.

For example, in a module <your_app>/selectors.py, we can have the following:

from sqlalchemy.orm import Session

def user_list(fetched_by: User, db_session: Session) -> Iterable[User]:
    user_ids = user_get_visible_for(user=fetched_by, db_session=db_session)

    query = User.id.in_(user_ids)

    return db_session.query(User).filter(query).all()

As you can see, user_get_visible_for is another selector.

You can return model, lists or whatever makes sense to your specific case.

Testing

Since services hold our business logic, they are an ideal candidate for tests.

If you decide to cover the service layer with tests, we have few general rules of thumb to follow:

The tests should cover the business logic in an exhaustive manner.
The tests should hit the database - creating & reading from it.
The tests should mock async task calls & everything that goes outside the project.

When creating the required state for a given test, one can use a combination of:

Fakes (we recommend using faker)
Other services, to create the required objects.
Special test utility & helper methods.
Factories (we recommend using factory_boy)

Let's take a look at our service from the example:

from sqlalchemy.orm import Session

from project.payment.exceptions import ItemAlreadyExists
from project.payments.services import payment_charge
from project.payments.selectors import items_get_for_user
from project.payments.models import Item, Payment


def item_buy(item: Item, user_id: str, db_session: Session) -> Payment:
    if item in items_get_for_user(user=user_id, db_session=db_session):
        raise ItemAlreadyExists

    with db_session.begin() as session:
        payment = Payment(
            item=item,
            user_id=user_id,
            successful=False
        )
        session.add(payment)

    payment_charge(payment_id=payment.id)

    return payment

The service:

Calls a selector for validation.
Creates an object.
Calls a service for payment.

Those are our tests:

import pytest
import uuid

from project.payment.exceptions import ItemAlreadyExists
from project.payments.services import item_buy
from project.payments.factories import ItemFactory
from project.payments.models import Payment, Item


class TestItemBuy:
    user_id = uuid.uuid4()

    @property
    def item(self) -> Item:
        return ItemFactory(user_id=self.user_id)

    def test_buying_item_that_is_already_bought_fails(self, mocker, test_db_session):
        """
        Since we already have tests for `items_get_for_user`,
        we can safely mock it here and give it a proper return value.
        """
        mocker.patch('project.payments.services.items_get_for_user', return_value=[self.item])
        with pytest.raises(ItemAlreadyExists):
            item_buy(user=self.user_id, item=self.item, db_session=test_db_session)

    def test_buying_item_creates_a_payment_and_calls_charge_service(self, mocker, test_db_session):
        """
        Since we already have tests for `payment_charge`,
        we can safely mock it here.
        """
        mocker.patch('project.payments.services.payment_charge')
        assert 0 == test_db_session.query(Payment).count()

        payment = item_buy(user=self.user_id, item=self.item, db_session=test_db_session)

        assert 1 == test_db_session.query(Payment).count()
        assert payment == test_db_session.query(Payment).first()
        assert payment.successful is False