Keep ORM Usage in Infrastructure Layer

ORM-specific code (entities, mappings, queries) belongs in the infrastructure layer. The domain and application layers should not know which ORM (or if any ORM) is being used.

Incorrect (ORM in application layer):

# application/usecases/get_orders.py
from sqlalchemy.orm import Session, joinedload
from sqlalchemy import and_, or_

class GetOrdersUseCase:
    def __init__(self, session: Session):  # SQLAlchemy session in use case
        self.session = session

    def execute(self, customer_id: str, filters: OrderFilters):
        # SQLAlchemy query in use case
        query = self.session.query(Order).options(
            joinedload(Order.items),
            joinedload(Order.customer)
        )

        if filters.status:
            query = query.filter(Order.status == filters.status)

        if filters.date_range:
            query = query.filter(and_(
                Order.created_at >= filters.date_range.start,
                Order.created_at <= filters.date_range.end
            ))

        return query.order_by(Order.created_at.desc()).all()

# Use case is now coupled to SQLAlchemy
# Switching to Peewee or raw SQL requires rewriting use case

Correct (ORM isolated in infrastructure):

# application/ports/order_repository.py
from abc import ABC, abstractmethod

class OrderRepository(ABC):
    @abstractmethod
    def find_by_customer(
        self,
        customer_id: CustomerId,
        filters: OrderFilters
    ) -> list[Order]:
        pass

# application/usecases/get_orders.py
class GetOrdersUseCase:
    def __init__(self, orders: OrderRepository):  # Interface, not Session
        self.orders = orders

    def execute(self, customer_id: str, filters: OrderFilters) -> list[Order]:
        return self.orders.find_by_customer(
            CustomerId(customer_id),
            filters
        )

# infrastructure/persistence/sqlalchemy_order_repository.py
from sqlalchemy.orm import Session, joinedload

class SqlAlchemyOrderRepository(OrderRepository):
    def __init__(self, session: Session):
        self.session = session

    def find_by_customer(
        self,
        customer_id: CustomerId,
        filters: OrderFilters
    ) -> list[Order]:
        query = self.session.query(OrderEntity).options(
            joinedload(OrderEntity.items)
        ).filter(OrderEntity.customer_id == customer_id.value)

        if filters.status:
            query = query.filter(OrderEntity.status == filters.status.value)

        entities = query.order_by(OrderEntity.created_at.desc()).all()
        return [self._to_domain(e) for e in entities]

    def _to_domain(self, entity: OrderEntity) -> Order:
        # Map ORM entity to domain entity
        pass

Benefits:

• Use case tests don’t need database or ORM setup
• Can switch ORMs without touching business logic
• Complex queries encapsulated in repository

Reference: Repository Pattern