From 0b167262285813937f1db1c6c0b6a096cc8cd410 Mon Sep 17 00:00:00 2001 From: acazares Date: Sat, 6 Dec 2025 23:37:13 -0600 Subject: [PATCH] feat: add electronic notices module with DTOs, models, routes, and service layer for managing electronic notices --- .../electronic_notices/__init__.py | 3 + .../electronic_notices/dto.py | 85 +++++++++ .../electronic_notices/models.py | 48 +++++ .../electronic_notices/routes.py | 154 +++++++++++++++ .../electronic_notices/service.py | 178 ++++++++++++++++++ 5 files changed, 468 insertions(+) create mode 100644 backend/api/v1/modules/a76/general_catalogs/electronic_notices/__init__.py create mode 100644 backend/api/v1/modules/a76/general_catalogs/electronic_notices/dto.py create mode 100644 backend/api/v1/modules/a76/general_catalogs/electronic_notices/models.py create mode 100644 backend/api/v1/modules/a76/general_catalogs/electronic_notices/routes.py create mode 100644 backend/api/v1/modules/a76/general_catalogs/electronic_notices/service.py diff --git a/backend/api/v1/modules/a76/general_catalogs/electronic_notices/__init__.py b/backend/api/v1/modules/a76/general_catalogs/electronic_notices/__init__.py new file mode 100644 index 00000000..12753c03 --- /dev/null +++ b/backend/api/v1/modules/a76/general_catalogs/electronic_notices/__init__.py @@ -0,0 +1,3 @@ +""" +Módulo de avisos electrónicos +""" diff --git a/backend/api/v1/modules/a76/general_catalogs/electronic_notices/dto.py b/backend/api/v1/modules/a76/general_catalogs/electronic_notices/dto.py new file mode 100644 index 00000000..523fad87 --- /dev/null +++ b/backend/api/v1/modules/a76/general_catalogs/electronic_notices/dto.py @@ -0,0 +1,85 @@ +""" +DTOs (Data Transfer Objects) para módulo de avisos electrónicos +""" + +from typing import Optional + +from pydantic import BaseModel, Field + + +class ElectronicNoticeCreateDTO(BaseModel): + """DTO para crear un aviso electrónico""" + + notice_number: Optional[str] = Field( + None, max_length=500, description="Notice number" + ) + year: Optional[str] = Field(None, max_length=20, description="Year") + patent: Optional[str] = Field(None, max_length=4, description="Patent") + pedimento: Optional[str] = Field( + None, max_length=15, description="Pedimento") + file_sent: Optional[str] = Field( + None, max_length=1000, description="File sent") + file_response: Optional[str] = Field( + None, max_length=1000, description="File response" + ) + status: Optional[str] = Field(None, max_length=100, description="Status") + invoice: Optional[str] = Field(None, max_length=50, description="Invoice") + validation_acknowledgment: Optional[str] = Field( + None, max_length=20, description="Validation acknowledgment" + ) + fea: Optional[str] = Field(None, max_length=1000, description="FEA") + certificate_number: Optional[str] = Field( + None, max_length=50, description="Certificate number" + ) + + class Config: + from_attributes = True + + +class ElectronicNoticeUpdateDTO(BaseModel): + """DTO para actualizar un aviso electrónico""" + + notice_number: Optional[str] = Field( + None, max_length=500, description="Notice number" + ) + year: Optional[str] = Field(None, max_length=20, description="Year") + patent: Optional[str] = Field(None, max_length=4, description="Patent") + pedimento: Optional[str] = Field( + None, max_length=15, description="Pedimento") + file_sent: Optional[str] = Field( + None, max_length=1000, description="File sent") + file_response: Optional[str] = Field( + None, max_length=1000, description="File response" + ) + status: Optional[str] = Field(None, max_length=100, description="Status") + invoice: Optional[str] = Field(None, max_length=50, description="Invoice") + validation_acknowledgment: Optional[str] = Field( + None, max_length=20, description="Validation acknowledgment" + ) + fea: Optional[str] = Field(None, max_length=1000, description="FEA") + certificate_number: Optional[str] = Field( + None, max_length=50, description="Certificate number" + ) + + class Config: + from_attributes = True + + +class ElectronicNoticeResponseDTO(BaseModel): + """DTO para responder con datos de un aviso electrónico""" + + sys_id: int + notice_number: Optional[str] = None + year: Optional[str] = None + patent: Optional[str] = None + pedimento: Optional[str] = None + file_sent: Optional[str] = None + file_response: Optional[str] = None + status: Optional[str] = None + invoice: Optional[str] = None + validation_acknowledgment: Optional[str] = None + fea: Optional[str] = None + certificate_number: Optional[str] = None + + class Config: + from_attributes = True diff --git a/backend/api/v1/modules/a76/general_catalogs/electronic_notices/models.py b/backend/api/v1/modules/a76/general_catalogs/electronic_notices/models.py new file mode 100644 index 00000000..9b4288a6 --- /dev/null +++ b/backend/api/v1/modules/a76/general_catalogs/electronic_notices/models.py @@ -0,0 +1,48 @@ +""" +Modelos ORM para gestión de avisos electrónicos +""" + +from typing import Optional + +from api.v1.common.base_models import TenantScopedMixin +from core.database import Base +from sqlalchemy import Integer, PrimaryKeyConstraint, String +from sqlalchemy.orm import Mapped, mapped_column + + +class ElectronicNotice(Base, TenantScopedMixin): + """ + Modelo para la tabla ElectronicNotice - Avisos Electrónicos + """ + + __tablename__ = "electronic_notices" # GAvisosElectronicos + __table_args__ = ( + PrimaryKeyConstraint("id", name="electronic_notices_pkey"), + {"schema": "a76"}, + ) + + # Primary key + id: Mapped[int] = mapped_column(Integer, primary_key=True) + + # Notice identification + notice_number: Mapped[Optional[str]] = mapped_column(String(500)) + year: Mapped[Optional[str]] = mapped_column(String(20)) + patent: Mapped[Optional[str]] = mapped_column(String(4)) + pedimento: Mapped[Optional[str]] = mapped_column(String(15)) + + # Files + file_sent: Mapped[Optional[str]] = mapped_column(String(1000)) + file_response: Mapped[Optional[str]] = mapped_column(String(1000)) + + # Status and validation + status: Mapped[Optional[str]] = mapped_column(String(100)) + invoice: Mapped[Optional[str]] = mapped_column(String(50)) + validation_acknowledgment: Mapped[Optional[str]] = mapped_column( + String(20)) + + # Certificate information + fea: Mapped[Optional[str]] = mapped_column(String(1000)) + certificate_number: Mapped[Optional[str]] = mapped_column(String(50)) + + def __repr__(self): + return f"" diff --git a/backend/api/v1/modules/a76/general_catalogs/electronic_notices/routes.py b/backend/api/v1/modules/a76/general_catalogs/electronic_notices/routes.py new file mode 100644 index 00000000..7d1a19e3 --- /dev/null +++ b/backend/api/v1/modules/a76/general_catalogs/electronic_notices/routes.py @@ -0,0 +1,154 @@ +""" +Rutas para gestión de avisos electrónicos +""" + +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 ( + ElectronicNoticeCreateDTO, + ElectronicNoticeResponseDTO, + ElectronicNoticeUpdateDTO, +) +from .models import ElectronicNotice +from .service import ElectronicNoticeService + +router = APIRouter(prefix="/electronic-notices", tags=["electronic-notices"]) + + +@router.get( + "", + response_model=dict, + summary="Get all electronic notices", +) +async def get_all_notices( + skip: int = Query(0, ge=0), + limit: int = Query(50, ge=1, le=100), + notice_number: str = Query(None), + status: str = Query(None), + pedimento: str = Query(None), + db: Session = Depends(get_core_db), +): + """Get all electronic notices with optional filtering and pagination""" + filters = {} + if notice_number: + filters["notice_number"] = notice_number + if status: + filters["status"] = status + if pedimento: + filters["pedimento"] = pedimento + + notices, total = ElectronicNoticeService.get_all(db, skip, limit, filters) + + return { + "data": [ + ElectronicNoticeResponseDTO.model_validate(notice) for notice in notices + ], + "total": total, + "skip": skip, + "limit": limit, + } + + +@router.get( + "/{sys_id}", + response_model=ElectronicNoticeResponseDTO, + summary="Get electronic notice by ID", +) +async def get_notice( + sys_id: int, + db: Session = Depends(get_core_db), +): + """Get an electronic notice by its ID""" + notice = ElectronicNoticeService.get_by_id(db, sys_id) + if not notice: + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail="Electronic notice not found", + ) + return ElectronicNoticeResponseDTO.model_validate(notice) + + +@router.post( + "", + response_model=ElectronicNoticeResponseDTO, + status_code=status.HTTP_201_CREATED, + summary="Create electronic notice", +) +async def create_notice( + notice_data: ElectronicNoticeCreateDTO, + db: Session = Depends(get_core_db), +): + """Create a new electronic notice""" + notice = ElectronicNoticeService.create(db, notice_data) + return ElectronicNoticeResponseDTO.model_validate(notice) + + +@router.put( + "/{sys_id}", + response_model=ElectronicNoticeResponseDTO, + summary="Update electronic notice", +) +async def update_notice( + sys_id: int, + notice_data: ElectronicNoticeUpdateDTO, + db: Session = Depends(get_core_db), +): + """Update an electronic notice""" + notice = ElectronicNoticeService.update(db, sys_id, notice_data) + if not notice: + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail="Electronic notice not found", + ) + return ElectronicNoticeResponseDTO.model_validate(notice) + + +@router.delete( + "/{sys_id}", + status_code=status.HTTP_204_NO_CONTENT, + summary="Delete electronic notice", +) +async def delete_notice( + sys_id: int, + db: Session = Depends(get_core_db), +): + """Delete an electronic notice""" + success = ElectronicNoticeService.delete(db, sys_id) + if not success: + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail="Electronic notice not found", + ) + return None + + +@router.get( + "/by-pedimento/{pedimento}", + response_model=List[ElectronicNoticeResponseDTO], + summary="Get notices by pedimento", +) +async def get_notices_by_pedimento( + pedimento: str, + db: Session = Depends(get_core_db), +): + """Get all electronic notices for a specific pedimento""" + notices = ElectronicNoticeService.get_by_pedimento(db, pedimento) + return [ElectronicNoticeResponseDTO.model_validate(notice) for notice in notices] + + +@router.get( + "/by-status/{status}", + response_model=List[ElectronicNoticeResponseDTO], + summary="Get notices by status", +) +async def get_notices_by_status( + status: str, + db: Session = Depends(get_core_db), +): + """Get all electronic notices with a specific status""" + notices = ElectronicNoticeService.get_by_status(db, status) + return [ElectronicNoticeResponseDTO.model_validate(notice) for notice in notices] diff --git a/backend/api/v1/modules/a76/general_catalogs/electronic_notices/service.py b/backend/api/v1/modules/a76/general_catalogs/electronic_notices/service.py new file mode 100644 index 00000000..e357fb78 --- /dev/null +++ b/backend/api/v1/modules/a76/general_catalogs/electronic_notices/service.py @@ -0,0 +1,178 @@ +""" +Capa de servicio para lógica de negocio de avisos electrónicos +""" + +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 ( + ElectronicNoticeCreateDTO, + ElectronicNoticeResponseDTO, + ElectronicNoticeUpdateDTO, +) +from .models import ElectronicNotice + +logger = logging.getLogger(__name__) + + +class ElectronicNoticeService: + """Servicio para gestión de avisos electrónicos""" + + 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[ElectronicNotice], int]: + """Get all electronic notices with pagination""" + query = db.query(ElectronicNotice) + + # Apply filters if provided + if filters: + if filters.get("notice_number"): + query = query.filter( + ElectronicNotice.notice_number.ilike( + f"%{filters['notice_number']}%" + ) + ) + if filters.get("status"): + query = query.filter( + ElectronicNotice.status.ilike(f"%{filters['status']}%") + ) + if filters.get("pedimento"): + query = query.filter( + ElectronicNotice.pedimento.ilike( + f"%{filters['pedimento']}%") + ) + + total = query.count() + notices = query.offset(skip).limit(limit).all() + + return notices, total + + @staticmethod + def get_by_id(db: Session, sys_id: int) -> Optional[ElectronicNotice]: + """Get electronic notice by ID""" + return db.query(ElectronicNotice).filter( + ElectronicNotice.sys_id == sys_id + ).first() + + @staticmethod + def create( + db: Session, notice_data: ElectronicNoticeCreateDTO + ) -> ElectronicNotice: + """Create a new electronic notice""" + try: + db_notice = ElectronicNotice( + **notice_data.model_dump(exclude_unset=True) + ) + + db.add(db_notice) + db.commit() + db.refresh(db_notice) + + return db_notice + + except IntegrityError as e: + db.rollback() + logger.error( + f"IntegrityError creating electronic notice: {str(e)}") + raise HTTPException( + status_code=400, + detail="Electronic notice already exists", + ) + except Exception as e: + db.rollback() + logger.error(f"Error creating electronic notice: {str(e)}") + raise HTTPException( + status_code=500, detail="Error creating electronic notice" + ) + + @staticmethod + def update( + db: Session, + sys_id: int, + notice_data: ElectronicNoticeUpdateDTO, + ) -> Optional[ElectronicNotice]: + """Update an electronic notice""" + try: + db_notice = db.query(ElectronicNotice).filter( + ElectronicNotice.sys_id == sys_id + ).first() + + if not db_notice: + return None + + for key, value in notice_data.model_dump(exclude_unset=True).items(): + setattr(db_notice, key, value) + + db.commit() + db.refresh(db_notice) + + return db_notice + + except IntegrityError as e: + db.rollback() + logger.error( + f"IntegrityError updating electronic notice: {str(e)}") + raise HTTPException( + status_code=400, + detail="Error updating electronic notice", + ) + except Exception as e: + db.rollback() + logger.error(f"Error updating electronic notice: {str(e)}") + raise HTTPException( + status_code=500, detail="Error updating electronic notice" + ) + + @staticmethod + def delete(db: Session, sys_id: int) -> bool: + """Delete an electronic notice""" + try: + db_notice = db.query(ElectronicNotice).filter( + ElectronicNotice.sys_id == sys_id + ).first() + + if not db_notice: + return False + + db.delete(db_notice) + db.commit() + + return True + + except Exception as e: + db.rollback() + logger.error(f"Error deleting electronic notice: {str(e)}") + raise HTTPException( + status_code=500, detail="Error deleting electronic notice" + ) + + @staticmethod + def get_by_pedimento( + db: Session, pedimento: str + ) -> List[ElectronicNotice]: + """Get all electronic notices by pedimento""" + return ( + db.query(ElectronicNotice) + .filter(ElectronicNotice.pedimento == pedimento) + .all() + ) + + @staticmethod + def get_by_status(db: Session, status: str) -> List[ElectronicNotice]: + """Get all electronic notices by status""" + return ( + db.query(ElectronicNotice) + .filter(ElectronicNotice.status == status) + .all() + )