feat: add error catalogs module with DTOs, models, routes, and service layer for managing error classifications and catalogs
This commit is contained in:
@@ -0,0 +1,3 @@
|
||||
"""
|
||||
Módulo de catálogos de errores
|
||||
"""
|
||||
@@ -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
|
||||
@@ -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"<ErrorClassification(id={self.id}, code={self.code}, level={self.level})>"
|
||||
|
||||
|
||||
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"<ErrorCatalog(id={self.id}, code={self.code}, description={self.description}, classification_id={self.classification_id})>"
|
||||
@@ -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]
|
||||
@@ -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")
|
||||
Reference in New Issue
Block a user