Repositories

Repositories form the data access layer of the application. Their purpose is to encapsulate all interactions with the database, providing a clean and consistent interface for querying, persisting, and deleting entities. Repositories do not contain business logic; instead, they focus solely on executing SQLAlchemy operations within the active session managed by the service layer.

In Atlas, each repository is implemented as a class that extends the base class Repository[T] for a given T entity. Each of its methods needs to be wrapped by the @repository_method decorator, which ensures that calls are properly integrated into the session stack and executed within the current transactional context.

Initializing the repositories package

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

Example:

src/repositories/__init__.py

from .product_repository import ProductRepository
from .category_repository import CategoryRepository
from .price_repository import PriceRepository
from .stock_repository import StockRepository
from .user_repository import UserRepository
from .admin_repository import AdminRepository
from .customer_repository import CustomerRepository

__all__ = [
    "ProductRepository",
    "CategoryRepository",
    "PriceRepository",
    "StockRepository",
    "UserRepository",
    "AdminRepository",
    "CustomerRepository"
]

Creating a Repository

To create a repository, follow the steps below:

• Create a file inside src/repositories with the name of the repository in snake_case;
• Declare the repository's class by extending Repository[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 repository's domain entity;
• For creating methods, wrap each of them with @repository_method;

Note: A repository has useful, generic default methods inherited from the base Repository[T] class. Check them out before implementing a method that sounds too generic.

src/repositories/customer_repository.py

from typing import List

from sqlalchemy import func, or_

from atlas.common import Page, Pageable
from atlas.decorators import repository_method
from atlas.layers import Repository

from models import Customer, User

class CustomerRepository(Repository[Customer]):
    @repository_method
    def list_by_criteria(
        self,
        search: str = None,
        uuids_to_exclude: List[str] = None,
        pageable: Pageable = Pageable()
    ) -> Page:
        query = self.model_query().join(Customer.user)
        if search:
            query = query.where(
                or_(
                    Customer.document.ilike(f"%{search}%"),
                    func.unaccent(User.name).ilike(func.unaccent(f"%{search}%")),
                    func.unaccent(User.username).ilike(func.unaccent(f"%{search}%")),
                    User.email.ilike(func.unaccent(f"%{search}%")),
                )
            )
        if uuids_to_exclude:
            query = query.where(Customer.uuid.notin_(uuids_to_exclude))
        return self.get_paginated_results(query, pageable)