Services

Services encapsulate the business logic of the application. Their purpose is to coordinate operations across repositories, enforce domain rules, and orchestrate the workflows that power each endpoint. Service methods run inside an active database session, allowing them to perform multiple repository operations within a single transactional context via SQLAlchemy.

In Atlas, each service is a class with each of its methods wrapped by the @service_method decorator for automatically stacking service calls. Each service can either be:

Domain-bound: associated with a domain by extending the base class Service[T] for a given T entity;
Domainless: defined as a plain class without extending Service[T], used for cross-cutting operations, integrations, utilities, or workflows that do not directly map to a specific domain entity.

Service Calls Stack

Atlas' services works with a service calls stack: when a service method is invoked, the service layer ensures that a database session exists. Nested service calls that comes after automatically reuse the same session, pushing onto the "service call stack" without opening additional connections.

Only the first (root) service call is responsible for committing or rolling back the transaction and closing the session. This guarantees transactional integrity across complex workflows while keeping database usage efficient and predictable.

Initializing the `services` package

Before creating services, create the src/services directory. This will be the Python package where all the API's services will be stored. Also, create a __init__.py file inside it for convenient importing.

Example:

src/services/__init__.py

from .product_service import ProductService
from .category_service import CategoryService
from .price_service import PriceService
from .stock_service import StockService
from .user_service import UserService
from .admin_service import AdminService
from .customer_service import CustomerService

__all__ = [
    "ProductService",
    "CategoryService",
    "PriceService",
    "StockService",
    "UserService",
    "AdminService",
    "CustomerService"
]

Creating a Domain-bound Service

To create a domain-bound service, follow the steps below:

Create a file inside src/services with the name of the service in snake_case;
Declare the service's class by extending Service[T] and naming it with the same name as the file, but in PascalCase — for the T type, use the model that will be bound as the service's domain entity;
Create a __init__ method for initializing the domain entity's repository and for dependency injection of other services, if needed;
For creating methods, wrap each of them with @service_method;

Note: A domain-bound service has useful, generic default methods inherited from the base Service[T] class (that interact with the domain's repository only). Check them out before implementing a method that sounds too generic.

Example:

src/services/customer_service.py

import bcrypt

from typing import List, Type

from atlas.common import ModelMapper, Page, Pageable
from atlas.decorators import service_method
from atlas.layers import Service, InputDTO, RetDTO

from models import Customer
from repositories import CustomerRepository

class CustomerService(Service[Customer]):
    def __init__(self) -> None:
        super().__init__()
        self.user_service = None
        self.customer_repository = CustomerRepository()

    @service_method
    def list_by_criteria(
        self,
        search: str = None,
        uuids_to_exclude: List[str] = None,
        pageable: Pageable = Pageable(),
        ret_dto_type: Type[RetDTO] = None
    ) -> RetDTO | Page:
        uuids_to_exclude = uuids_to_exclude or []
        page = self.customer_repository.list_by_criteria(search=search, uuids_to_exclude=uuids_to_exclude, pageable=pageable)
        return self.map_or_self(page, ret_dto_type)

    @service_method
    def create(self, input_dto_instance: InputDTO, ret_dto_type: Type[RetDTO] = None) -> RetDTO | Customer:
        self.validate_uniqueness_by_fields(input_dto_instance, ["document"])
        self.user_service.validate_uniqueness_by_fields(input_dto_instance.user, ["email", "username"])

        input_dto_instance.user.password = bcrypt.hashpw(input_dto_instance.user.password.encode(), bcrypt.gensalt()).decode()

        customer = ModelMapper.map_instance_to_type(input_dto_instance, Customer)
        customer = self.customer_repository.add(customer)
        return self.map_or_self(customer, ret_dto_type)

    @service_method
    def update(self, uuid: str, input_dto_instance: InputDTO, ret_dto_type: Type[RetDTO] = None) -> RetDTO | Customer:
        existing_customer = self.get_by_uuid(uuid)

        self.validate_uniqueness_by_fields(input_dto_instance, ["document"], uuid_to_exclude=existing_customer.uuid)
        if input_dto_instance.user:
            self.user_service.validate_uniqueness_by_fields(input_dto_instance.user, ["email", "username"], uuid_to_exclude=existing_customer.user.uuid)
            if input_dto_instance.user.password:
                input_dto_instance.user.password = bcrypt.hashpw(input_dto_instance.user.password.encode(), bcrypt.gensalt()).decode()
            input_dto_instance.user.uuid = existing_customer.user.uuid

        ModelMapper.map(input_dto_instance, existing_customer)
        return self.map_or_self(existing_customer, ret_dto_type)

Creating a Domainless Service

To create a domainless service, follow the steps below:

Create a file inside src/services with the name of the service in snake_case;
Declare the service's class as a simple class (without heritage) and name it with the same name as the file, but in PascalCase;
For creating methods, wrap each of them with @service_method;

Example:

src/integration/client/aws/ses_service.py

from typing import TypeVar
from atlas.decorators import service_method
from atlas.dto import DTO
from atlas.exceptions import IntegrationError

InputDTO = TypeVar("InputDTO", bound=DTO)

class SESService:
    @service_method
    def send_contact(self, dto: InputDTO) -> None:
        try:
            body_text = (
                f"New contact received:\n"
                f"Name: {dto.name}\n"
                f"Email: {dto.email}\n"
                f"Company: {dto.company}\n"
                f"Phone: {dto.phone}\n"
                f"Message: {dto.message}\n"
            )

            body_html = f"""
                <h3>New contact received</h3>
                <p><b>Name:</b> {dto.name}</p>
                <p><b>Email:</b> {dto.email}</p>
                <p><b>Company:</b> {dto.company}</p>
                <p><b>Phone:</b> {dto.phone}</p>
                <p><b>Message:</b><br>{dto.message}</p>
            """

            import boto3
            ses = boto3.client("ses")
            ses.send_email(
                Source="Mydal <contact@mydal.com.br>",
                Destination={"ToAddresses": ["mydal.software@gmail.com"]},
                Message={
                    "Subject": {"Data": "New Contact"},
                    "Body": {"Text": {"Data": body_text}, "Html": {"Data": body_html}},
                },
                ReplyToAddresses=[dto.email]
            )

        except Exception as e:
            raise IntegrationError(str(e))

Dependency Injection

For importing services into endpoints in the controller layer or into other services, it’s up to the developer to decide how to structure the implementation. However, the recommended approach is to implement a dependency injection mechanism, such as the one described below.

src/services/__di_container__.py

# Do all business layer dependency injections here
from typing import Any, TYPE_CHECKING

from .user_service import UserService
from .admin_service import AdminService
from .customer_service import CustomerService
from .address_service import AddressService

# Private internal container to manage instances and dependencies
class _LazyContainer:
    _user_service = None
    _admin_service = None
    _customer_service = None
    _address_service = None

    @property
    def user_service(self):
        if self._user_service is None:
            self._user_service = UserService()
            self._user_service.permission_service = self.permission_service # Declared at UserService's __init__ method
        return self._user_service

    @property
    def admin_service(self):
        if self._admin_service is None:
            self._admin_service = AdminService()
            self._admin_service.user_service = self.user_service # Declared at AdminService's __init__ method
        return self._admin_service

    @property
    def customer_service(self):
        if self._customer_service is None:
            self._customer_service = CustomerService()
            self._customer_service.user_service = self.user_service # Declared at CustomerService's __init__ method
        return self._customer_service

    @property
    def address_service(self):
        if self._address_service is None:
            self._address_service = AddressService()
            self._address_service.customer_service = self.customer_service # Declared at AddressService's __init__ method
        return self._address_service

_container = _LazyContainer()

def __getattr__(name: str) -> Any:
    if hasattr(_container, name):
        return getattr(_container, name)
    raise AttributeError(f"module \"{__name__}\" has no attribute \"{name}\"")

if TYPE_CHECKING:
    user_service: UserService
    admin_service: AdminService
    customer_service: CustomerService
    address_service: AddressService

Importing service inside another service:

src/services/customer_service.py

import bcrypt

from typing import List, Type

from atlas.common import ModelMapper, Page, Pageable
from atlas.decorators import service_method
from atlas.layers import Service, InputDTO, RetDTO

from models import Customer
from repositories import CustomerRepository

class CustomerService(Service[Customer]):
    def __init__(self) -> None:
        super().__init__()
        self.user_service = None # This will be filled with the injected service via __di_container__
        self.customer_repository = CustomerRepository()

    ...

Importing service in an endpoint:

src/functions/manager/login/POST.py

import os
import sys
from typing import Dict, Any

sys.path.append(os.path.dirname(os.path.abspath(__file__)))
from .__resources__.post_request_dto import RequestDTO
from .__resources__.post_response_dto import ResponseDTO

from atlas.decorators import controller_method
from atlas.http import HttpRequest, HttpResponse

from common import Messages
from services.__di_container__ import user_service

@controller_method
def lambda_handler(req: HttpRequest) -> Dict[str, Any]:
    request_dto = req.body(RequestDTO)

    token_dict = user_service.login(request_dto.username, request_dto.password)

    return HttpResponse.build(200, ResponseDTO(
        message=Messages.get("SUCCESSFULLY_LOGGED_IN"),
        **token_dict
    ).model_dump_json())

Conventions

There are some conventions for how services interact with repositories.

A service should only directly call its domain's own repository, if domain-bound — e.g. CustomerRepository can only have its methods called by CustomerService.
Access from a service to other repository bound to a different domain should be possible only through its correspondent service — e.g. AdminService should not access CustomerRepository; instead, it should access CustomerRepository's functions/features/resources through CustomerService (CustomerRepository's domain-bound, related service).

Services

Service Calls Stack

Initializing the services package

Creating a Domain-bound Service

Creating a Domainless Service

Dependency Injection

Conventions

Initializing the `services` package