feat: add prevalidators module with DTOs, models, routes, and service layer for managing prevalidators

backend/api/v1/modules/a76/general_catalogs/prevalidators/__init__.py

@@ -0,0 +1,3 @@
+"""
+Módulo de prevalidadores
+"""

backend/api/v1/modules/a76/general_catalogs/prevalidators/dto.py

@@ -0,0 +1,87 @@
+"""
+DTOs (Data Transfer Objects) para módulo de prevalidadores
+"""
+
+from typing import Optional
+
+from pydantic import BaseModel, Field
+
+
+class PrevalidatorCreateDTO(BaseModel):
+    """DTO para crear un prevalidador"""
+
+    code: str = Field(..., max_length=20, description="Prevalidator code")
+    customs_prevalidator: Optional[str] = Field(
+        None, max_length=20, description="Customs prevalidator"
+    )
+    patent_prevalidator: Optional[str] = Field(
+        None, max_length=20, description="Patent prevalidator"
+    )
+    description: Optional[str] = Field(
+        None, max_length=50, description="Description"
+    )
+
+    class Config:
+        from_attributes = True
+
+
+class PrevalidatorUpdateDTO(BaseModel):
+    """DTO para actualizar un prevalidador"""
+
+    code: Optional[str] = Field(
+        None, max_length=20, description="Prevalidator code"
+    )
+    customs_prevalidator: Optional[str] = Field(
+        None, max_length=20, description="Customs prevalidator"
+    )
+    patent_prevalidator: Optional[str] = Field(
+        None, max_length=20, description="Patent prevalidator"
+    )
+    description: Optional[str] = Field(
+        None, max_length=50, description="Description"
+    )
+
+    class Config:
+        from_attributes = True
+
+
+class PrevalidatorResponseDTO(BaseModel):
+    """DTO para responder con datos de un prevalidador"""
+
+    id: int
+    code: str
+    customs_prevalidator: Optional[str] = None
+    patent_prevalidator: Optional[str] = None
+    description: Optional[str] = None
+
+    class Config:
+        from_attributes = True
+
+
+class PrevalidatorUpdateDTO(BaseModel):
+    """DTO para actualizar un prevalidador"""
+
+    customs_prevalidator: Optional[str] = Field(
+        None, max_length=20, description="Customs prevalidator"
+    )
+    patent_prevalidator: Optional[str] = Field(
+        None, max_length=20, description="Patent prevalidator"
+    )
+    description: Optional[str] = Field(
+        None, max_length=50, description="Description"
+    )
+
+    class Config:
+        from_attributes = True
+
+
+class PrevalidatorResponseDTO(BaseModel):
+    """DTO para responder con datos de un prevalidador"""
+
+    code: str
+    customs_prevalidator: Optional[str] = None
+    patent_prevalidator: Optional[str] = None
+    description: Optional[str] = None
+
+    class Config:
+        from_attributes = True

backend/api/v1/modules/a76/general_catalogs/prevalidators/models.py

@@ -0,0 +1,37 @@
+"""
+Modelos ORM para gestión de prevalidadores
+"""
+
+from typing import Optional
+
+from api.v1.common.base_models import TenantScopedMixin
+from core.database import Base
+from sqlalchemy import Integer, PrimaryKeyConstraint, String, UniqueConstraint
+from sqlalchemy.orm import Mapped, mapped_column
+
+
+class Prevalidator(Base, TenantScopedMixin):
+    """
+    Modelo para la tabla Prevalidator - Prevalidadores
+    """
+
+    __tablename__ = "prevalidators"  # GPrevalidadores
+    __table_args__ = (
+        PrimaryKeyConstraint("id", name="prevalidators_pkey"),
+        UniqueConstraint("code", name="prevalidators_code_unique"),
+        {"schema": "a76"},
+    )
+
+    # Primary key
+    id: Mapped[int] = mapped_column(Integer, primary_key=True)
+
+    # Prevalidator code (unique)
+    code: Mapped[str] = mapped_column(String(20), nullable=False, unique=True)
+
+    # Prevalidator information
+    customs_prevalidator: Mapped[Optional[str]] = mapped_column(String(20))
+    patent_prevalidator: Mapped[Optional[str]] = mapped_column(String(20))
+    description: Mapped[Optional[str]] = mapped_column(String(50))
+
+    def __repr__(self):
+        return f"<Prevalidator(code={self.code}, description={self.description})>"

backend/api/v1/modules/a76/general_catalogs/prevalidators/routes.py

@@ -0,0 +1,165 @@
+"""
+Rutas para gestión de prevalidadores
+"""
+
+from typing import List
+
+from fastapi import APIRouter, Depends, HTTPException, Query, status
+from sqlalchemy.orm import Session
+
+from core.database import get_core_db
+from .dto import (
+    PrevalidatorCreateDTO,
+    PrevalidatorResponseDTO,
+    PrevalidatorUpdateDTO,
+)
+from .models import Prevalidator
+from .service import PrevalidatorService
+
+router = APIRouter(prefix="/prevalidators", tags=["prevalidators"])
+
+
+@router.get(
+    "",
+    response_model=dict,
+    summary="Get all prevalidators",
+)
+async def get_all_prevalidators(
+    skip: int = Query(0, ge=0),
+    limit: int = Query(50, ge=1, le=100),
+    code: str = Query(None),
+    description: str = Query(None),
+    customs_prevalidator: str = Query(None),
+    db: Session = Depends(get_core_db),
+):
+    """Get all prevalidators with optional filtering and pagination"""
+    filters = {}
+    if code:
+        filters["code"] = code
+    if description:
+        filters["description"] = description
+    if customs_prevalidator:
+        filters["customs_prevalidator"] = customs_prevalidator
+
+    prevalidators, total = PrevalidatorService.get_all(
+        db, skip, limit, filters)
+
+    return {
+        "data": [
+            PrevalidatorResponseDTO.model_validate(prevalidator)
+            for prevalidator in prevalidators
+        ],
+        "total": total,
+        "skip": skip,
+        "limit": limit,
+    }
+
+
+@router.get(
+    "/{prevalidator_id}",
+    response_model=PrevalidatorResponseDTO,
+    summary="Get prevalidator by ID",
+)
+async def get_prevalidator(
+    prevalidator_id: int,
+    db: Session = Depends(get_core_db),
+):
+    """Get a prevalidator by its ID"""
+    prevalidator = PrevalidatorService.get_by_id(db, prevalidator_id)
+    if not prevalidator:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail="Prevalidator not found",
+        )
+    return PrevalidatorResponseDTO.model_validate(prevalidator)
+
+
+@router.get(
+    "/code/{code}",
+    response_model=PrevalidatorResponseDTO,
+    summary="Get prevalidator by code",
+)
+async def get_prevalidator_by_code(
+    code: str,
+    db: Session = Depends(get_core_db),
+):
+    """Get a prevalidator by its code"""
+    prevalidator = PrevalidatorService.get_by_code(db, code)
+    if not prevalidator:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail="Prevalidator not found",
+        )
+    return PrevalidatorResponseDTO.model_validate(prevalidator)
+
+
+@router.post(
+    "",
+    response_model=PrevalidatorResponseDTO,
+    status_code=status.HTTP_201_CREATED,
+    summary="Create prevalidator",
+)
+async def create_prevalidator(
+    prevalidator_data: PrevalidatorCreateDTO,
+    db: Session = Depends(get_core_db),
+):
+    """Create a new prevalidator"""
+    prevalidator = PrevalidatorService.create(db, prevalidator_data)
+    return PrevalidatorResponseDTO.model_validate(prevalidator)
+
+
+@router.put(
+    "/{prevalidator_id}",
+    response_model=PrevalidatorResponseDTO,
+    summary="Update prevalidator",
+)
+async def update_prevalidator(
+    prevalidator_id: int,
+    prevalidator_data: PrevalidatorUpdateDTO,
+    db: Session = Depends(get_core_db),
+):
+    """Update a prevalidator"""
+    prevalidator = PrevalidatorService.update(
+        db, prevalidator_id, prevalidator_data)
+    if not prevalidator:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail="Prevalidator not found",
+        )
+    return PrevalidatorResponseDTO.model_validate(prevalidator)
+
+
+@router.delete(
+    "/{prevalidator_id}",
+    status_code=status.HTTP_204_NO_CONTENT,
+    summary="Delete prevalidator",
+)
+async def delete_prevalidator(
+    prevalidator_id: int,
+    db: Session = Depends(get_core_db),
+):
+    """Delete a prevalidator"""
+    success = PrevalidatorService.delete(db, prevalidator_id)
+    if not success:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail="Prevalidator not found",
+        )
+    return None
+
+
+@router.get(
+    "/by-customs/{customs}",
+    response_model=List[PrevalidatorResponseDTO],
+    summary="Get prevalidators by customs",
+)
+async def get_prevalidators_by_customs(
+    customs: str,
+    db: Session = Depends(get_core_db),
+):
+    """Get all prevalidators for a specific customs"""
+    prevalidators = PrevalidatorService.get_by_customs(db, customs)
+    return [
+        PrevalidatorResponseDTO.model_validate(prevalidator)
+        for prevalidator in prevalidators
+    ]

backend/api/v1/modules/a76/general_catalogs/prevalidators/service.py

@@ -0,0 +1,164 @@
+"""
+Capa de servicio para lógica de negocio de prevalidadores
+"""
+
+import logging
+from typing import Any, Dict, List, Optional, Tuple
+
+from fastapi import HTTPException
+from sqlalchemy.exc import IntegrityError
+from sqlalchemy.orm import Session
+
+from .dto import (
+    PrevalidatorCreateDTO,
+    PrevalidatorResponseDTO,
+    PrevalidatorUpdateDTO,
+)
+from .models import Prevalidator
+
+logger = logging.getLogger(__name__)
+
+
+class PrevalidatorService:
+    """Servicio para gestión de prevalidadores"""
+
+    def __init__(self, db: Session):
+        self.db = db
+
+    @staticmethod
+    def get_all(
+        db: Session,
+        skip: int = 0,
+        limit: int = 50,
+        filters: Optional[Dict[str, Any]] = None,
+    ) -> Tuple[List[Prevalidator], int]:
+        """Get all prevalidators with pagination"""
+        query = db.query(Prevalidator)
+
+        # Apply filters if provided
+        if filters:
+            if filters.get("code"):
+                query = query.filter(
+                    Prevalidator.code.ilike(f"%{filters['code']}%")
+                )
+            if filters.get("description"):
+                query = query.filter(
+                    Prevalidator.description.ilike(
+                        f"%{filters['description']}%")
+                )
+            if filters.get("customs_prevalidator"):
+                query = query.filter(
+                    Prevalidator.customs_prevalidator.ilike(
+                        f"%{filters['customs_prevalidator']}%"
+                    )
+                )
+
+        total = query.count()
+        prevalidators = query.offset(skip).limit(limit).all()
+
+        return prevalidators, total
+
+    @staticmethod
+    def get_by_code(db: Session, code: str) -> Optional[Prevalidator]:
+        """Get prevalidator by code"""
+        return db.query(Prevalidator).filter(Prevalidator.code == code).first()
+
+    @staticmethod
+    def get_by_id(db: Session, prevalidator_id: int) -> Optional[Prevalidator]:
+        """Get prevalidator by ID"""
+        return db.query(Prevalidator).filter(Prevalidator.id == prevalidator_id).first()
+
+    @staticmethod
+    def create(db: Session, prevalidator_data: PrevalidatorCreateDTO) -> Prevalidator:
+        """Create a new prevalidator"""
+        try:
+            db_prevalidator = Prevalidator(
+                **prevalidator_data.model_dump(exclude_unset=True)
+            )
+
+            db.add(db_prevalidator)
+            db.commit()
+            db.refresh(db_prevalidator)
+
+            return db_prevalidator
+
+        except IntegrityError as e:
+            db.rollback()
+            logger.error(f"IntegrityError creating prevalidator: {str(e)}")
+            raise HTTPException(
+                status_code=400,
+                detail="Prevalidator already exists",
+            )
+        except Exception as e:
+            db.rollback()
+            logger.error(f"Error creating prevalidator: {str(e)}")
+            raise HTTPException(
+                status_code=500, detail="Error creating prevalidator")
+
+    @staticmethod
+    def update(
+        db: Session,
+        prevalidator_id: int,
+        prevalidator_data: PrevalidatorUpdateDTO,
+    ) -> Optional[Prevalidator]:
+        """Update a prevalidator"""
+        try:
+            db_prevalidator = db.query(Prevalidator).filter(
+                Prevalidator.id == prevalidator_id
+            ).first()
+
+            if not db_prevalidator:
+                return None
+
+            for key, value in prevalidator_data.model_dump(exclude_unset=True).items():
+                setattr(db_prevalidator, key, value)
+
+            db.commit()
+            db.refresh(db_prevalidator)
+
+            return db_prevalidator
+
+        except IntegrityError as e:
+            db.rollback()
+            logger.error(f"IntegrityError updating prevalidator: {str(e)}")
+            raise HTTPException(
+                status_code=400,
+                detail="Error updating prevalidator",
+            )
+        except Exception as e:
+            db.rollback()
+            logger.error(f"Error updating prevalidator: {str(e)}")
+            raise HTTPException(
+                status_code=500, detail="Error updating prevalidator"
+            )
+
+    @staticmethod
+    def delete(db: Session, prevalidator_id: int) -> bool:
+        """Delete a prevalidator"""
+        try:
+            db_prevalidator = db.query(Prevalidator).filter(
+                Prevalidator.id == prevalidator_id
+            ).first()
+
+            if not db_prevalidator:
+                return False
+
+            db.delete(db_prevalidator)
+            db.commit()
+
+            return True
+
+        except Exception as e:
+            db.rollback()
+            logger.error(f"Error deleting prevalidator: {str(e)}")
+            raise HTTPException(
+                status_code=500, detail="Error deleting prevalidator")
+
+    @staticmethod
+    def get_by_customs(db: Session, customs: str) -> List[Prevalidator]:
+        """Get all prevalidators by customs"""
+        return (
+            db.query(Prevalidator)
+            .filter(Prevalidator.customs_prevalidator == customs)
+            .all()
+        )