from flask import flash, redirect, render_template, url_forfrom app.exceptions import ConflictErrorfrom .forms import ColorFormfrom .services import ColorService@colors_bp.route('/create', methods=['GET', 'POST'])def create_color(): """ Muestra el formulario y crea un nuevo color en el catálogo. GET: Renderiza el formulario de creación. POST: Valida el formulario, crea el color y redirige. """ form = ColorForm() if form.validate_on_submit(): data = {'name': form.name.data} try: ColorService.create(data) flash('Color creado exitosamente', 'success') return redirect(url_for('colors.create_color')) except ConflictError as e: flash(e.message, 'error') return render_template('colors/create.html', form=form)
from sqlalchemy import funcfrom sqlalchemy.exc import IntegrityErrorfrom app.exceptions import ConflictError, ValidationError, NotFoundErrorfrom app.extensions import dbfrom app.models.color import Colorclass ColorService: """Servicio para operaciones de negocio relacionadas con colores.""" @staticmethod def create(data: dict) -> dict: """ Crea un nuevo color en el catálogo. Args: data: Diccionario con los datos del color (name requerido) Returns: dict: Color creado serializado Raises: ValidationError: Si el nombre está vacío o no se proporciona ConflictError: Si ya existe un color con el mismo nombre """ name = data.get('name') if not name or not name.strip(): raise ValidationError('El nombre del color es requerido') name = name.strip() existing = Color.query.filter( func.lower(Color.name) == name.lower() ).first() if existing: raise ConflictError(f"Ya existe un color con el nombre '{name}'") color = Color(name=name) db.session.add(color) try: db.session.commit() except IntegrityError: db.session.rollback() raise ConflictError(f"Ya existe un color con el nombre '{name}'") return color.to_dict() @staticmethod def get_by_id(id_color: int) -> Color: """Obtiene un color por su ID.""" color = Color.query.get(id_color) if not color: raise NotFoundError(f"No se encontró un color con ID {id_color}") return color @staticmethod def update(id_color: int, data: dict) -> dict: """Actualiza un color existente.""" color = ColorService.get_by_id(id_color) name = data.get('name') if not name or not name.strip(): raise ValidationError('El nombre del color es requerido') name = name.strip() existing = ( db.session.query(Color.id_color) .filter( func.lower(Color.name) == name.lower(), Color.id_color != id_color ) .first() is not None ) if existing: raise ConflictError(f"Ya existe otro color con el nombre '{name}'") color.name = name try: db.session.commit() except IntegrityError: db.session.rollback() raise ConflictError(f"Ya existe otro color con el nombre '{name}'") return color.to_dict() @staticmethod def delete(id_color: int) -> None: """Elimina un color (soft delete).""" color = ColorService.get_by_id(id_color) color.active = False color.deleted_at = func.current_timestamp() db.session.commit()
from sqlalchemy.sql import funcfrom ..extensions import dbclass Color(db.Model): """ Modelo de Color para catálogo de referencias de colores. Attributes: id_color: Identificador único del color. name: Nombre del color. active: Indica si el color está activo o no. created_at: Fecha de creación del color. updated_at: Fecha de última actualización del color. deleted_at: Fecha de eliminación lógica del color. created_by: Usuario que creó el color. updated_by: Usuario que actualizó el color. deleted_by: Usuario que eliminó el color. """ __tablename__ = 'colors' id_color = db.Column(db.Integer, primary_key=True) name = db.Column(db.String(50), nullable=False, unique=True) active = db.Column(db.Boolean, nullable=False, default=True) created_at = db.Column( db.TIMESTAMP, nullable=False, server_default=func.current_timestamp() ) updated_at = db.Column( db.TIMESTAMP, nullable=False, server_default=func.current_timestamp(), server_onupdate=func.current_timestamp() ) deleted_at = db.Column(db.TIMESTAMP, nullable=True) created_by = db.Column(db.String(100), nullable=True) updated_by = db.Column(db.String(100), nullable=True) deleted_by = db.Column(db.String(100), nullable=True) def to_dict(self) -> dict: """ Serializa el modelo a diccionario. Returns: dict: Representación del color en formato diccionario """ return { 'id_color': self.id_color, 'name': self.name, 'active': self.active, 'created_at': self.created_at.isoformat() if self.created_at else None, 'updated_at': self.updated_at.isoformat() if self.updated_at else None, }
Here’s a complete flow showing how all three layers work together:
# 1. PRESENTATION LAYER - Route handles the request@colors_bp.route('/create', methods=['POST'])def create_color(): form = ColorForm() if form.validate_on_submit(): data = {'name': form.name.data} try: # 2. Delegate to SERVICE LAYER ColorService.create(data) flash('Color creado exitosamente', 'success') return redirect(url_for('colors.list_colors')) except ConflictError as e: flash(e.message, 'error')# 2. SERVICE LAYER - Implements business logicclass ColorService: @staticmethod def create(data: dict) -> dict: name = data.get('name') # Business validation if not name or not name.strip(): raise ValidationError('El nombre del color es requerido') # Check for duplicates existing = Color.query.filter_by(name=name).first() if existing: raise ConflictError(f"Ya existe un color con el nombre '{name}'") # 3. Use MODEL LAYER to persist data color = Color(name=name) db.session.add(color) db.session.commit() return color.to_dict()# 3. MODEL LAYER - Defines the entity structureclass Color(db.Model): __tablename__ = 'colors' id_color = db.Column(db.Integer, primary_key=True) name = db.Column(db.String(50), nullable=False, unique=True) active = db.Column(db.Boolean, nullable=False, default=True)