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

2025-12-06 23:37:13 -06:00
parent 3fdfd70633
commit 0b16726228
5 changed files with 468 additions and 0 deletions
--- a/backend/api/v1/modules/a76/general_catalogs/electronic_notices/init.py
+++ b/backend/api/v1/modules/a76/general_catalogs/electronic_notices/init.py
@@ -0,0 +1,3 @@
+"""
+Módulo de avisos electrónicos
+"""
--- a/backend/api/v1/modules/a76/general_catalogs/electronic_notices/dto.py
+++ 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
--- a/backend/api/v1/modules/a76/general_catalogs/electronic_notices/models.py
+++ 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"<ElectronicNotice(id={self.id}, notice_number={self.notice_number}, status={self.status})>"
--- a/backend/api/v1/modules/a76/general_catalogs/electronic_notices/routes.py
+++ 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]
--- a/backend/api/v1/modules/a76/general_catalogs/electronic_notices/service.py
+++ 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()
+        )