diff --git a/backend/api/v1/modules/a76/general_catalogs/error_catalogs/__init__.py b/backend/api/v1/modules/a76/general_catalogs/error_catalogs/__init__.py new file mode 100644 index 00000000..4537d3f3 --- /dev/null +++ b/backend/api/v1/modules/a76/general_catalogs/error_catalogs/__init__.py @@ -0,0 +1,3 @@ +""" +Módulo de catálogos de errores +""" diff --git a/backend/api/v1/modules/a76/general_catalogs/error_catalogs/dto.py b/backend/api/v1/modules/a76/general_catalogs/error_catalogs/dto.py new file mode 100644 index 00000000..fd8e5033 --- /dev/null +++ b/backend/api/v1/modules/a76/general_catalogs/error_catalogs/dto.py @@ -0,0 +1,105 @@ +""" +DTOs (Data Transfer Objects) para módulo de catálogos de errores +""" + +from typing import Optional, List + +from pydantic import BaseModel, Field + + +# ============ ERROR CLASSIFICATION DTOS ============ +class ErrorClassificationCreateDTO(BaseModel): + """DTO para crear una clasificación de error""" + + code: str = Field(..., max_length=100, description="Classification code") + level: Optional[str] = Field(None, max_length=3, description="Level") + + class Config: + from_attributes = True + + +class ErrorClassificationUpdateDTO(BaseModel): + """DTO para actualizar una clasificación de error""" + + level: Optional[str] = Field(None, max_length=3, description="Level") + + class Config: + from_attributes = True + + +class ErrorClassificationResponseDTO(BaseModel): + """DTO para responder con datos de una clasificación de error""" + + id: int + code: str + level: Optional[str] = None + + class Config: + from_attributes = True + + +class ErrorClassificationDetailResponseDTO(BaseModel): + """DTO detallado para responder con datos de una clasificación y sus errores""" + + id: int + code: str + level: Optional[str] = None + errors: List["ErrorCatalogResponseDTO"] = [] + + class Config: + from_attributes = True + + +# ============ ERROR CATALOG DTOS ============ +class ErrorCatalogCreateDTO(BaseModel): + """DTO para crear un error en el catálogo""" + + code: str = Field(..., max_length=15, description="Error code") + description: Optional[str] = Field( + None, max_length=255, description="Error description" + ) + classification_id: Optional[int] = Field( + None, description="Classification ID" + ) + + class Config: + from_attributes = True + + +class ErrorCatalogUpdateDTO(BaseModel): + """DTO para actualizar un error en el catálogo""" + + description: Optional[str] = Field( + None, max_length=255, description="Error description" + ) + classification_id: Optional[int] = Field( + None, description="Classification ID" + ) + + class Config: + from_attributes = True + + +class ErrorCatalogResponseDTO(BaseModel): + """DTO para responder con datos de un error en el catálogo""" + + id: int + code: str + description: Optional[str] = None + classification_id: Optional[int] = None + + class Config: + from_attributes = True + + +class ErrorCatalogDetailResponseDTO(BaseModel): + """DTO detallado para responder con datos de un error y su clasificación""" + + id: int + code: str + description: Optional[str] = None + classification_id: Optional[int] = None + classification: Optional[ErrorClassificationResponseDTO] = None + + class Config: + from_attributes = True diff --git a/backend/api/v1/modules/a76/general_catalogs/error_catalogs/models.py b/backend/api/v1/modules/a76/general_catalogs/error_catalogs/models.py new file mode 100644 index 00000000..eca4ab85 --- /dev/null +++ b/backend/api/v1/modules/a76/general_catalogs/error_catalogs/models.py @@ -0,0 +1,80 @@ +""" +Modelos ORM para gestión de catálogos de errores +""" + +from typing import Optional + +from api.v1.common.base_models import TenantScopedMixin +from core.database import Base +from sqlalchemy import ForeignKeyConstraint, Integer, PrimaryKeyConstraint, String, UniqueConstraint +from sqlalchemy.orm import Mapped, mapped_column, relationship + + +class ErrorClassification(Base, TenantScopedMixin): + """ + Modelo para la tabla ErrorClassification - Clasificación de Errores + """ + + __tablename__ = "error_classifications" # GCatErroresClas + __table_args__ = ( + PrimaryKeyConstraint("id", name="error_classifications_pkey"), + UniqueConstraint("code", name="error_classifications_code_unique"), + {"schema": "a76"}, + ) + + # Primary key + id: Mapped[int] = mapped_column( + Integer, primary_key=True, autoincrement=True) + + # Classification code (unique) + code: Mapped[str] = mapped_column(String(100), nullable=False, unique=True) + + # Classification information + level: Mapped[Optional[str]] = mapped_column(String(3)) + + # Relationships + errors: Mapped[list["ErrorCatalog"]] = relationship( + "ErrorCatalog", back_populates="classification", cascade="all, delete-orphan" + ) + + def __repr__(self): + return f"" + + +class ErrorCatalog(Base, TenantScopedMixin): + """ + Modelo para la tabla ErrorCatalog - Catálogo de Errores + """ + + __tablename__ = "error_catalogs" # GCatErrores + __table_args__ = ( + PrimaryKeyConstraint("id", name="error_catalogs_pkey"), + UniqueConstraint("code", name="error_catalogs_code_unique"), + ForeignKeyConstraint( + ["classification_id"], + ["a76.error_classifications.id"], + name="fk_error_catalogs_classification", + ), + {"schema": "a76"}, + ) + + # Primary key + id: Mapped[int] = mapped_column( + Integer, primary_key=True, autoincrement=True) + + # Error code (unique) + code: Mapped[str] = mapped_column(String(15), nullable=False, unique=True) + + # Error information + description: Mapped[Optional[str]] = mapped_column(String(255)) + + # Foreign key to classification + classification_id: Mapped[Optional[int]] = mapped_column(Integer) + + # Relationships + classification: Mapped[Optional["ErrorClassification"]] = relationship( + "ErrorClassification", back_populates="errors" + ) + + def __repr__(self): + return f"" diff --git a/backend/api/v1/modules/a76/general_catalogs/error_catalogs/routes.py b/backend/api/v1/modules/a76/general_catalogs/error_catalogs/routes.py new file mode 100644 index 00000000..1116b5a5 --- /dev/null +++ b/backend/api/v1/modules/a76/general_catalogs/error_catalogs/routes.py @@ -0,0 +1,295 @@ +""" +Rutas para gestión de catálogos de errores +""" + +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 ( + ErrorClassificationCreateDTO, + ErrorClassificationResponseDTO, + ErrorClassificationUpdateDTO, + ErrorClassificationDetailResponseDTO, + ErrorCatalogCreateDTO, + ErrorCatalogResponseDTO, + ErrorCatalogUpdateDTO, + ErrorCatalogDetailResponseDTO, +) +from .models import ErrorClassification, ErrorCatalog +from .service import ErrorClassificationService, ErrorCatalogService + +router = APIRouter(prefix="/error-catalogs", tags=["error-catalogs"]) + + +# ============ ERROR CLASSIFICATIONS ENDPOINTS ============ +@router.get( + "/classifications", + response_model=dict, + summary="Get all error classifications", +) +async def get_all_classifications( + skip: int = Query(0, ge=0), + limit: int = Query(50, ge=1, le=100), + code: str = Query(None), + level: str = Query(None), + db: Session = Depends(get_core_db), +): + """Get all error classifications with optional filtering and pagination""" + filters = {} + if code: + filters["code"] = code + if level: + filters["level"] = level + + classifications, total = ErrorClassificationService.get_all( + db, skip, limit, filters + ) + + return { + "data": [ + ErrorClassificationResponseDTO.model_validate(classification) + for classification in classifications + ], + "total": total, + "skip": skip, + "limit": limit, + } + + +@router.get( + "/classifications/{classification_id}", + response_model=ErrorClassificationDetailResponseDTO, + summary="Get error classification by ID with errors", +) +async def get_classification( + classification_id: int, + db: Session = Depends(get_core_db), +): + """Get an error classification by its ID with all related errors""" + classification = ErrorClassificationService.get_by_id( + db, classification_id) + if not classification: + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail="Classification not found", + ) + return ErrorClassificationDetailResponseDTO.model_validate(classification) + + +@router.get( + "/classifications/code/{code}", + response_model=ErrorClassificationDetailResponseDTO, + summary="Get error classification by code with errors", +) +async def get_classification_by_code( + code: str, + db: Session = Depends(get_core_db), +): + """Get an error classification by its code with all related errors""" + classification = ErrorClassificationService.get_by_code(db, code) + if not classification: + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail="Classification not found", + ) + return ErrorClassificationDetailResponseDTO.model_validate(classification) + + +@router.post( + "/classifications", + response_model=ErrorClassificationResponseDTO, + status_code=status.HTTP_201_CREATED, + summary="Create error classification", +) +async def create_classification( + classification_data: ErrorClassificationCreateDTO, + db: Session = Depends(get_core_db), +): + """Create a new error classification""" + classification = ErrorClassificationService.create(db, classification_data) + return ErrorClassificationResponseDTO.model_validate(classification) + + +@router.put( + "/classifications/{classification_id}", + response_model=ErrorClassificationResponseDTO, + summary="Update error classification", +) +async def update_classification( + classification_id: int, + classification_data: ErrorClassificationUpdateDTO, + db: Session = Depends(get_core_db), +): + """Update an error classification""" + classification = ErrorClassificationService.update( + db, classification_id, classification_data + ) + if not classification: + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail="Classification not found", + ) + return ErrorClassificationResponseDTO.model_validate(classification) + + +@router.delete( + "/classifications/{classification_id}", + status_code=status.HTTP_204_NO_CONTENT, + summary="Delete error classification", +) +async def delete_classification( + classification_id: int, + db: Session = Depends(get_core_db), +): + """Delete an error classification""" + success = ErrorClassificationService.delete(db, classification_id) + if not success: + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail="Classification not found", + ) + return None + + +# ============ ERROR CATALOG ENDPOINTS ============ +@router.get( + "", + response_model=dict, + summary="Get all errors in catalog", +) +async def get_all_errors( + skip: int = Query(0, ge=0), + limit: int = Query(50, ge=1, le=100), + code: str = Query(None), + description: str = Query(None), + classification_code: str = Query(None), + db: Session = Depends(get_core_db), +): + """Get all error catalogs with optional filtering and pagination""" + filters = {} + if code: + filters["code"] = code + if description: + filters["description"] = description + if classification_code: + filters["classification_code"] = classification_code + + catalogs, total = ErrorCatalogService.get_all(db, skip, limit, filters) + + return { + "data": [ + ErrorCatalogResponseDTO.model_validate(catalog) for catalog in catalogs + ], + "total": total, + "skip": skip, + "limit": limit, + } + + +@router.get( + "/{error_id}", + response_model=ErrorCatalogDetailResponseDTO, + summary="Get error by ID", +) +async def get_error( + error_id: int, + db: Session = Depends(get_core_db), +): + """Get an error by its ID with classification details""" + error = ErrorCatalogService.get_by_id(db, error_id) + if not error: + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail="Error not found", + ) + return ErrorCatalogDetailResponseDTO.model_validate(error) + + +@router.get( + "/code/{code}", + response_model=ErrorCatalogDetailResponseDTO, + summary="Get error by code", +) +async def get_error_by_code( + code: str, + db: Session = Depends(get_core_db), +): + """Get an error by its code with classification details""" + error = ErrorCatalogService.get_by_code(db, code) + if not error: + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail="Error not found", + ) + return ErrorCatalogDetailResponseDTO.model_validate(error) + + +@router.post( + "", + response_model=ErrorCatalogResponseDTO, + status_code=status.HTTP_201_CREATED, + summary="Create error in catalog", +) +async def create_error( + error_data: ErrorCatalogCreateDTO, + db: Session = Depends(get_core_db), +): + """Create a new error in the catalog""" + error = ErrorCatalogService.create(db, error_data) + return ErrorCatalogResponseDTO.model_validate(error) + + +@router.put( + "/{error_id}", + response_model=ErrorCatalogResponseDTO, + summary="Update error in catalog", +) +async def update_error( + error_id: int, + error_data: ErrorCatalogUpdateDTO, + db: Session = Depends(get_core_db), +): + """Update an error in the catalog""" + error = ErrorCatalogService.update(db, error_id, error_data) + if not error: + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail="Error not found", + ) + return ErrorCatalogResponseDTO.model_validate(error) + + +@router.delete( + "/{error_id}", + status_code=status.HTTP_204_NO_CONTENT, + summary="Delete error from catalog", +) +async def delete_error( + error_id: int, + db: Session = Depends(get_core_db), +): + """Delete an error from the catalog""" + success = ErrorCatalogService.delete(db, error_id) + if not success: + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail="Error not found", + ) + return None + + +@router.get( + "/classification/{classification_id}", + response_model=List[ErrorCatalogResponseDTO], + summary="Get errors by classification", +) +async def get_errors_by_classification( + classification_id: int, + db: Session = Depends(get_core_db), +): + """Get all errors for a specific classification""" + errors = ErrorCatalogService.get_by_classification(db, classification_id) + return [ErrorCatalogResponseDTO.model_validate(error) for error in errors] diff --git a/backend/api/v1/modules/a76/general_catalogs/error_catalogs/service.py b/backend/api/v1/modules/a76/general_catalogs/error_catalogs/service.py new file mode 100644 index 00000000..8cddc486 --- /dev/null +++ b/backend/api/v1/modules/a76/general_catalogs/error_catalogs/service.py @@ -0,0 +1,336 @@ +""" +Capa de servicio para lógica de negocio de catálogos de errores +""" + +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 ( + ErrorClassificationCreateDTO, + ErrorClassificationResponseDTO, + ErrorClassificationUpdateDTO, + ErrorCatalogCreateDTO, + ErrorCatalogResponseDTO, + ErrorCatalogUpdateDTO, +) +from .models import ErrorClassification, ErrorCatalog + +logger = logging.getLogger(__name__) + + +class ErrorClassificationService: + """Servicio para gestión de clasificaciones de errores""" + + 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[ErrorClassification], int]: + """Get all error classifications with pagination""" + query = db.query(ErrorClassification) + + if filters: + if filters.get("code"): + query = query.filter( + ErrorClassification.code.ilike(f"%{filters['code']}%") + ) + if filters.get("level"): + query = query.filter( + ErrorClassification.level.ilike(f"%{filters['level']}%") + ) + + total = query.count() + classifications = query.offset(skip).limit(limit).all() + + return classifications, total + + @staticmethod + def get_by_code(db: Session, code: str) -> Optional[ErrorClassification]: + """Get error classification by code""" + return ( + db.query(ErrorClassification) + .filter(ErrorClassification.code == code) + .first() + ) + + @staticmethod + def get_by_id(db: Session, classification_id: int) -> Optional[ErrorClassification]: + """Get error classification by ID""" + return ( + db.query(ErrorClassification) + .filter(ErrorClassification.id == classification_id) + .first() + ) + + @staticmethod + def create( + db: Session, classification_data: ErrorClassificationCreateDTO + ) -> ErrorClassification: + """Create a new error classification""" + try: + db_classification = ErrorClassification( + **classification_data.model_dump(exclude_unset=True) + ) + + db.add(db_classification) + db.commit() + db.refresh(db_classification) + + return db_classification + + except IntegrityError as e: + db.rollback() + logger.error( + f"IntegrityError creating error classification: {str(e)}") + raise HTTPException( + status_code=400, + detail="Error classification already exists", + ) + except Exception as e: + db.rollback() + logger.error(f"Error creating error classification: {str(e)}") + raise HTTPException( + status_code=500, detail="Error creating error classification" + ) + + @staticmethod + def update( + db: Session, + classification_id: int, + classification_data: ErrorClassificationUpdateDTO, + ) -> Optional[ErrorClassification]: + """Update an error classification""" + try: + db_classification = ( + db.query(ErrorClassification) + .filter(ErrorClassification.id == classification_id) + .first() + ) + + if not db_classification: + return None + + for key, value in classification_data.model_dump(exclude_unset=True).items(): + setattr(db_classification, key, value) + + db.commit() + db.refresh(db_classification) + + return db_classification + + except IntegrityError as e: + db.rollback() + logger.error( + f"IntegrityError updating error classification: {str(e)}") + raise HTTPException( + status_code=400, + detail="Error updating error classification", + ) + except Exception as e: + db.rollback() + logger.error(f"Error updating error classification: {str(e)}") + raise HTTPException( + status_code=500, detail="Error updating error classification" + ) + + @staticmethod + def delete(db: Session, classification_id: int) -> bool: + """Delete an error classification""" + try: + db_classification = ( + db.query(ErrorClassification) + .filter(ErrorClassification.id == classification_id) + .first() + ) + + if not db_classification: + return False + + db.delete(db_classification) + db.commit() + + return True + + except Exception as e: + db.rollback() + logger.error(f"Error deleting error classification: {str(e)}") + raise HTTPException( + status_code=500, detail="Error deleting error classification" + ) + + +class ErrorCatalogService: + """Servicio para gestión de catálogos de errores""" + + 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[ErrorCatalog], int]: + """Get all error catalogs with pagination""" + query = db.query(ErrorCatalog) + + if filters: + if filters.get("code"): + query = query.filter( + ErrorCatalog.code.ilike(f"%{filters['code']}%")) + if filters.get("description"): + query = query.filter( + ErrorCatalog.description.ilike( + f"%{filters['description']}%") + ) + if filters.get("classification_id"): + query = query.filter( + ErrorCatalog.classification_id == filters['classification_id'] + ) + + total = query.count() + catalogs = query.offset(skip).limit(limit).all() + + return catalogs, total + + @staticmethod + def get_by_code(db: Session, code: str) -> Optional[ErrorCatalog]: + """Get error catalog by code""" + return db.query(ErrorCatalog).filter(ErrorCatalog.code == code).first() + + @staticmethod + def get_by_id(db: Session, error_id: int) -> Optional[ErrorCatalog]: + """Get error catalog by ID""" + return db.query(ErrorCatalog).filter(ErrorCatalog.id == error_id).first() + + @staticmethod + def get_by_classification( + db: Session, classification_id: int + ) -> List[ErrorCatalog]: + """Get all errors by classification""" + return ( + db.query(ErrorCatalog) + .filter(ErrorCatalog.classification_id == classification_id) + .all() + ) + + @staticmethod + def create(db: Session, error_data: ErrorCatalogCreateDTO) -> ErrorCatalog: + """Create a new error catalog""" + try: + # Validate classification exists if provided + if error_data.classification_id: + classification = ( + db.query(ErrorClassification) + .filter(ErrorClassification.id == error_data.classification_id) + .first() + ) + if not classification: + raise HTTPException( + status_code=400, + detail="Classification not found", + ) + + db_error = ErrorCatalog( + **error_data.model_dump(exclude_unset=True)) + + db.add(db_error) + db.commit() + db.refresh(db_error) + + return db_error + + except HTTPException: + raise + except IntegrityError as e: + db.rollback() + logger.error(f"IntegrityError creating error catalog: {str(e)}") + raise HTTPException( + status_code=400, + detail="Error already exists", + ) + except Exception as e: + db.rollback() + logger.error(f"Error creating error catalog: {str(e)}") + raise HTTPException( + status_code=500, detail="Error creating error catalog") + + @staticmethod + def update( + db: Session, error_id: int, error_data: ErrorCatalogUpdateDTO + ) -> Optional[ErrorCatalog]: + """Update an error catalog""" + try: + # Validate classification exists if provided + if error_data.classification_id: + classification = ( + db.query(ErrorClassification) + .filter(ErrorClassification.id == error_data.classification_id) + .first() + ) + if not classification: + raise HTTPException( + status_code=400, + detail="Classification not found", + ) + + db_error = db.query(ErrorCatalog).filter( + ErrorCatalog.id == error_id).first() + + if not db_error: + return None + + for key, value in error_data.model_dump(exclude_unset=True).items(): + setattr(db_error, key, value) + + db.commit() + db.refresh(db_error) + + return db_error + + except HTTPException: + raise + except IntegrityError as e: + db.rollback() + logger.error(f"IntegrityError updating error catalog: {str(e)}") + raise HTTPException( + status_code=400, + detail="Error updating error catalog", + ) + except Exception as e: + db.rollback() + logger.error(f"Error updating error catalog: {str(e)}") + raise HTTPException( + status_code=500, detail="Error updating error catalog" + ) + + @staticmethod + def delete(db: Session, error_id: int) -> bool: + """Delete an error catalog""" + try: + db_error = db.query(ErrorCatalog).filter( + ErrorCatalog.id == error_id).first() + + if not db_error: + return False + + db.delete(db_error) + db.commit() + + return True + + except Exception as e: + db.rollback() + logger.error(f"Error deleting error catalog: {str(e)}") + raise HTTPException( + status_code=500, detail="Error deleting error catalog")