Service Layer en FastAPI

Separación de responsabilidades

El Service Layer representa una capa intermedia fundamental en aplicaciones FastAPI que se sitúa entre los endpoints de la API y la lógica de acceso a datos. Esta arquitectura permite mantener una separación clara de responsabilidades, donde cada componente tiene un propósito específico y bien definido.

La separación de responsabilidades en FastAPI mediante el Service Layer se basa en el principio de que cada capa debe tener una responsabilidad única y claramente definida. Los endpoints se encargan únicamente de manejar las peticiones HTTP, los servicios contienen la lógica de negocio, y los repositories gestionan el acceso a datos.

Estructura de capas en FastAPI

Una aplicación bien estructurada con Service Layer mantiene la siguiente jerarquía de responsabilidades:

• Capa de presentación (endpoints): Maneja peticiones HTTP, validación de entrada y formateo de respuestas
• Capa de servicio: Implementa la lógica de negocio y coordina operaciones
• Capa de repositorio: Gestiona el acceso y persistencia de datos
• Capa de modelo: Define las entidades de dominio

Esta estructura permite que cada capa se enfoque en su responsabilidad específica sin conocer los detalles de implementación de otras capas.

Implementación básica del Service Layer

Un servicio en FastAPI encapsula la lógica de negocio relacionada con una entidad o funcionalidad específica. Aquí tienes un ejemplo básico de implementación:

# services/user_service.py
from typing import Optional, List
from repositories.user_repository import UserRepository
from schemas.user_schemas import UserCreate, UserUpdate
from models.user import User

class UserService:
    def __init__(self, user_repository: UserRepository):
        self.user_repository = user_repository
    
    def create_user(self, user_data: UserCreate) -> User:
        # Lógica de negocio antes de crear
        if self._email_exists(user_data.email):
            raise ValueError("El email ya está en uso")
        
        # Aplicar reglas de negocio
        processed_data = self._process_user_data(user_data)
        
        # Delegar la persistencia al repository
        return self.user_repository.create(processed_data)
    
    def _email_exists(self, email: str) -> bool:
        """Validación de regla de negocio privada"""
        existing_user = self.user_repository.find_by_email(email)
        return existing_user is not None
    
    def _process_user_data(self, user_data: UserCreate) -> dict:
        """Procesar datos según reglas de negocio"""
        return {
            "name": user_data.name.strip().title(),
            "email": user_data.email.lower(),
            "is_active": True
        }

Separación de concerns en endpoints

Los endpoints de FastAPI se simplifican considerablemente cuando utilizan el Service Layer, ya que solo necesitan manejar aspectos relacionados con HTTP:

# routes/users.py
from fastapi import APIRouter, Depends, HTTPException
from services.user_service import UserService
from schemas.user_schemas import UserCreate, UserResponse
from dependencies import get_user_service

router = APIRouter()

@router.post("/users/", response_model=UserResponse)
async def create_user(
    user_data: UserCreate,
    user_service: UserService = Depends(get_user_service)
):
    try:
        # El endpoint solo maneja HTTP concerns
        user = user_service.create_user(user_data)
        return user
    except ValueError as e:
        # Traducir excepciones de negocio a HTTP
        raise HTTPException(status_code=400, detail=str(e))

Beneficios de la separación de responsabilidades

La separación clara de responsabilidades mediante Service Layer aporta múltiples ventajas a tu aplicación FastAPI:

• Mantenibilidad: Cada capa es independiente y puede modificarse sin afectar otras
• Testabilidad: Los servicios pueden probarse de forma aislada sin depender de la infraestructura HTTP
• Reutilización: La lógica de negocio puede reutilizarse en diferentes contextos
• Escalabilidad: Permite dividir el trabajo entre diferentes desarrolladores por capas

Gestión de dependencias entre capas

Para mantener la separación adecuada, es importante gestionar las dependencias correctamente. FastAPI permite inyectar servicios en los endpoints:

# dependencies.py
from fastapi import Depends
from sqlalchemy.orm import Session
from database import get_db
from repositories.user_repository import UserRepository
from services.user_service import UserService

def get_user_repository(db: Session = Depends(get_db)) -> UserRepository:
    return UserRepository(db)

def get_user_service(
    user_repo: UserRepository = Depends(get_user_repository)
) -> UserService:
    return UserService(user_repo)

Esta aproximación garantiza que cada capa reciba sus dependencias de forma controlada, manteniendo el acoplamiento bajo y la cohesión alta entre componentes relacionados.

Composición de servicios

La composición de servicios permite crear funcionalidades complejas combinando múltiples servicios especializados. Este patrón es especialmente útil cuando una operación de negocio requiere coordinar diferentes dominios o entidades dentro de tu aplicación FastAPI.

Principios de composición

Un servicio compuesto actúa como coordinador entre varios servicios especializados, orquestando sus operaciones para completar un flujo de negocio complejo. La clave está en mantener cada servicio enfocado en su dominio específico mientras permite su colaboración.

La composición horizontal ocurre cuando un servicio utiliza múltiples servicios del mismo nivel para completar una operación:

# services/order_service.py
from typing import List
from services.user_service import UserService
from services.product_service import ProductService
from services.inventory_service import InventoryService
from services.payment_service import PaymentService
from schemas.order_schemas import OrderCreate, OrderResponse

class OrderService:
    def __init__(
        self,
        user_service: UserService,
        product_service: ProductService,
        inventory_service: InventoryService,
        payment_service: PaymentService
    ):
        self.user_service = user_service
        self.product_service = product_service
        self.inventory_service = inventory_service
        self.payment_service = payment_service
    
    def process_order(self, order_data: OrderCreate) -> OrderResponse:
        # Coordinar múltiples servicios especializados
        user = self.user_service.validate_user(order_data.user_id)
        products = self.product_service.get_products(order_data.product_ids)
        
        # Verificar disponibilidad antes del procesamiento
        self.inventory_service.reserve_stock(order_data.items)
        
        try:
            # Procesar pago
            payment = self.payment_service.process_payment(
                user_id=user.id,
                amount=self._calculate_total(products, order_data.items)
            )
            
            # Confirmar la orden solo si el pago es exitoso
            return self._create_confirmed_order(user, products, payment)
            
        except PaymentError:
            # Liberar stock reservado en caso de fallo
            self.inventory_service.release_reserved_stock(order_data.items)
            raise

Patrón de servicios agregados

Los servicios agregados encapsulan operaciones que requieren coordinar múltiples entidades relacionadas. Este patrón es útil cuando necesitas mantener consistencia entre diferentes dominios:

# services/ecommerce_service.py
class EcommerceService:
    def __init__(
        self,
        order_service: OrderService,
        notification_service: NotificationService,
        audit_service: AuditService
    ):
        self.order_service = order_service
        self.notification_service = notification_service
        self.audit_service = audit_service
    
    def complete_purchase(self, purchase_data: PurchaseRequest) -> PurchaseResult:
        # Orquestar el flujo completo de compra
        order = self.order_service.process_order(purchase_data.order)
        
        # Registrar la operación para auditoría
        self.audit_service.log_purchase(order.id, purchase_data.user_id)
        
        # Notificar al usuario y al sistema
        self.notification_service.send_order_confirmation(
            user_id=purchase_data.user_id,
            order_id=order.id
        )
        
        return PurchaseResult(
            order=order,
            status="completed",
            notifications_sent=True
        )

Composición con inyección de dependencias

FastAPI facilita la composición de servicios mediante su sistema de inyección de dependencias, permitiendo crear jerarquías complejas de servicios de forma limpia:

# dependencies.py
from fastapi import Depends

def get_order_service(
    user_service: UserService = Depends(get_user_service),
    product_service: ProductService = Depends(get_product_service),
    inventory_service: InventoryService = Depends(get_inventory_service),
    payment_service: PaymentService = Depends(get_payment_service)
) -> OrderService:
    return OrderService(
        user_service,
        product_service, 
        inventory_service,
        payment_service
    )

def get_ecommerce_service(
    order_service: OrderService = Depends(get_order_service),
    notification_service: NotificationService = Depends(get_notification_service),
    audit_service: AuditService = Depends(get_audit_service)
) -> EcommerceService:
    return EcommerceService(order_service, notification_service, audit_service)

Manejo de transacciones en servicios compuestos

Cuando múltiples servicios participan en una operación, es crucial manejar las transacciones de forma coordinada. Una aproximación común es usar el patrón Unit of Work:

# services/transaction_service.py
from contextlib import contextmanager
from sqlalchemy.orm import Session

class TransactionalService:
    def __init__(self, db_session: Session):
        self.db = db_session
    
    @contextmanager
    def transaction(self):
        """Context manager para operaciones transaccionales"""
        try:
            yield self.db
            self.db.commit()
        except Exception:
            self.db.rollback()
            raise
        finally:
            self.db.close()

class CompositeOrderService(OrderService):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
    
    def process_order_transactionally(self, order_data: OrderCreate) -> OrderResponse:
        with self.transaction() as tx:
            # Todas las operaciones dentro de la misma transacción
            user = self.user_service.validate_user(order_data.user_id)
            
            # Reservar inventario
            reservation = self.inventory_service.reserve_with_transaction(
                order_data.items, tx
            )
            
            # Procesar pago
            payment = self.payment_service.process_with_transaction(
                user.id, self._calculate_total(order_data.items), tx
            )
            
            # Crear orden final
            return self._create_order_with_transaction(
                user, reservation, payment, tx
            )

Composición condicional

Los servicios pueden componer dinámicamente otros servicios basándose en condiciones específicas del negocio:

# services/flexible_order_service.py
class FlexibleOrderService:
    def __init__(
        self,
        standard_payment_service: PaymentService,
        premium_payment_service: PremiumPaymentService,
        user_service: UserService
    ):
        self.standard_payment = standard_payment_service
        self.premium_payment = premium_payment_service
        self.user_service = user_service
    
    def process_adaptive_order(self, order_data: OrderCreate) -> OrderResponse:
        user = self.user_service.get_user(order_data.user_id)
        
        # Seleccionar servicio de pago basado en el tipo de usuario
        payment_service = (
            self.premium_payment if user.is_premium 
            else self.standard_payment
        )
        
        # Aplicar lógica específica según el contexto
        if user.is_premium:
            return self._process_premium_order(order_data, payment_service)
        else:
            return self._process_standard_order(order_data, payment_service)

Comunicación entre servicios compuestos

Para mantener el bajo acoplamiento, los servicios compuestos deben comunicarse a través de interfaces bien definidas y evitar el conocimiento directo de implementaciones internas:

# services/catalog_service.py
from abc import ABC, abstractmethod
from typing import Protocol

class ProductPricingProtocol(Protocol):
    def calculate_price(self, product_id: int, quantity: int) -> float: ...

class CatalogService:
    def __init__(
        self,
        product_service: ProductService,
        pricing_service: ProductPricingProtocol,  # Interface, no implementación
        recommendation_service: RecommendationService
    ):
        self.product_service = product_service
        self.pricing_service = pricing_service
        self.recommendation_service = recommendation_service
    
    def get_product_catalog(self, user_id: int, category: str) -> CatalogResponse:
        # Obtener productos base
        products = self.product_service.get_by_category(category)
        
        # Enriquecer con precios dinámicos
        enriched_products = []
        for product in products:
            price = self.pricing_service.calculate_price(product.id, 1)
            enriched_products.append({
                **product.model_dump(),
                "current_price": price
            })
        
        # Agregar recomendaciones personalizadas
        recommendations = self.recommendation_service.get_for_user(user_id)
        
        return CatalogResponse(
            products=enriched_products,
            recommendations=recommendations,
            category=category
        )

Esta aproximación de composición de servicios permite crear aplicaciones FastAPI escalables y mantenibles, donde cada servicio mantiene su responsabilidad específica mientras colabora efectivamente con otros para completar operaciones de negocio complejas.