diff --git a/backend/api/v1/modules/a76/general_catalogs/signatures/__init__.py b/backend/api/v1/modules/a76/general_catalogs/signatures/__init__.py new file mode 100644 index 00000000..4ee29dba --- /dev/null +++ b/backend/api/v1/modules/a76/general_catalogs/signatures/__init__.py @@ -0,0 +1,3 @@ +""" +Módulo de firmas +""" diff --git a/backend/api/v1/modules/a76/general_catalogs/signatures/dto.py b/backend/api/v1/modules/a76/general_catalogs/signatures/dto.py new file mode 100644 index 00000000..4524f91c --- /dev/null +++ b/backend/api/v1/modules/a76/general_catalogs/signatures/dto.py @@ -0,0 +1,44 @@ +""" +DTOs (Data Transfer Objects) para módulo de firmas +""" + +from typing import Optional + +from pydantic import BaseModel, Field + + +class SignatureCreateDTO(BaseModel): + """DTO para crear una firma""" + + code: str = Field(..., max_length=10, description="Signature code") + signature: Optional[str] = Field( + None, max_length=1000, description="Signature") + photo_path: Optional[str] = Field( + None, max_length=1000, description="Photo path") + + class Config: + from_attributes = True + + +class SignatureUpdateDTO(BaseModel): + """DTO para actualizar una firma""" + + signature: Optional[str] = Field( + None, max_length=1000, description="Signature") + photo_path: Optional[str] = Field( + None, max_length=1000, description="Photo path") + + class Config: + from_attributes = True + + +class SignatureResponseDTO(BaseModel): + """DTO para responder con datos de una firma""" + + id: int + code: str + signature: Optional[str] = None + photo_path: Optional[str] = None + + class Config: + from_attributes = True diff --git a/backend/api/v1/modules/a76/general_catalogs/signatures/models.py b/backend/api/v1/modules/a76/general_catalogs/signatures/models.py new file mode 100644 index 00000000..8a0fddb6 --- /dev/null +++ b/backend/api/v1/modules/a76/general_catalogs/signatures/models.py @@ -0,0 +1,37 @@ +""" +Modelos ORM para gestión de firmas +""" + +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 Signature(Base, TenantScopedMixin): + """ + Modelo para la tabla Signature - Firmas + """ + + __tablename__ = "signatures" # GFirmas + __table_args__ = ( + PrimaryKeyConstraint("id", name="signatures_pkey"), + UniqueConstraint("code", name="signatures_code_unique"), + {"schema": "a76"}, + ) + + # Primary key + id: Mapped[int] = mapped_column( + Integer, primary_key=True, autoincrement=True) + + # Signature code (unique) + code: Mapped[str] = mapped_column(String(10), nullable=False, unique=True) + + # Signature information + signature: Mapped[Optional[str]] = mapped_column(String(1000)) + photo_path: Mapped[Optional[str]] = mapped_column(String(1000)) + + def __repr__(self): + return f"" diff --git a/backend/api/v1/modules/a76/general_catalogs/signatures/routes.py b/backend/api/v1/modules/a76/general_catalogs/signatures/routes.py new file mode 100644 index 00000000..9ac77be0 --- /dev/null +++ b/backend/api/v1/modules/a76/general_catalogs/signatures/routes.py @@ -0,0 +1,131 @@ +""" +Rutas para gestión de firmas +""" + +from fastapi import APIRouter, Depends, HTTPException, Query, status +from sqlalchemy.orm import Session + +from core.database import get_core_db +from .dto import SignatureCreateDTO, SignatureResponseDTO, SignatureUpdateDTO +from .models import Signature +from .service import SignatureService + +router = APIRouter(prefix="/signatures", tags=["signatures"]) + + +@router.get( + "", + response_model=dict, + summary="Get all signatures", +) +async def get_all_signatures( + skip: int = Query(0, ge=0), + limit: int = Query(50, ge=1, le=100), + code: str = Query(None), + db: Session = Depends(get_core_db), +): + """Get all signatures with optional filtering and pagination""" + filters = {} + if code: + filters["code"] = code + + signatures, total = SignatureService.get_all(db, skip, limit, filters) + + return { + "data": [SignatureResponseDTO.model_validate(sig) for sig in signatures], + "total": total, + "skip": skip, + "limit": limit, + } + + +@router.get( + "/{signature_id}", + response_model=SignatureResponseDTO, + summary="Get signature by ID", +) +async def get_signature( + signature_id: int, + db: Session = Depends(get_core_db), +): + """Get a signature by its ID""" + signature = SignatureService.get_by_id(db, signature_id) + if not signature: + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail="Signature not found", + ) + return SignatureResponseDTO.model_validate(signature) + + +@router.get( + "/code/{code}", + response_model=SignatureResponseDTO, + summary="Get signature by code", +) +async def get_signature_by_code( + code: str, + db: Session = Depends(get_core_db), +): + """Get a signature by its code""" + signature = SignatureService.get_by_code(db, code) + if not signature: + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail="Signature not found", + ) + return SignatureResponseDTO.model_validate(signature) + + +@router.post( + "", + response_model=SignatureResponseDTO, + status_code=status.HTTP_201_CREATED, + summary="Create signature", +) +async def create_signature( + signature_data: SignatureCreateDTO, + db: Session = Depends(get_core_db), +): + """Create a new signature""" + signature = SignatureService.create(db, signature_data) + return SignatureResponseDTO.model_validate(signature) + + +@router.put( + "/{signature_id}", + response_model=SignatureResponseDTO, + summary="Update signature", +) +async def update_signature( + signature_id: int, + signature_data: SignatureUpdateDTO, + db: Session = Depends(get_core_db), +): + """Update a signature""" + signature = SignatureService.update(db, signature_id, signature_data) + if not signature: + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail="Signature not found", + ) + return SignatureResponseDTO.model_validate(signature) + + +@router.delete( + "/{signature_id}", + status_code=status.HTTP_204_NO_CONTENT, + summary="Delete signature", +) +async def delete_signature( + signature_id: int, + db: Session = Depends(get_core_db), +): + """Delete a signature""" + success = SignatureService.delete(db, signature_id) + if not success: + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail="Signature not found", + ) + return None diff --git a/backend/api/v1/modules/a76/general_catalogs/signatures/service.py b/backend/api/v1/modules/a76/general_catalogs/signatures/service.py new file mode 100644 index 00000000..f278a635 --- /dev/null +++ b/backend/api/v1/modules/a76/general_catalogs/signatures/service.py @@ -0,0 +1,134 @@ +""" +Capa de servicio para lógica de negocio de firmas +""" + +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 SignatureCreateDTO, SignatureResponseDTO, SignatureUpdateDTO +from .models import Signature + +logger = logging.getLogger(__name__) + + +class SignatureService: + """Servicio para gestión de firmas""" + + 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[Signature], int]: + """Get all signatures with pagination""" + query = db.query(Signature) + + if filters: + if filters.get("code"): + query = query.filter( + Signature.code.ilike(f"%{filters['code']}%")) + + total = query.count() + signatures = query.offset(skip).limit(limit).all() + + return signatures, total + + @staticmethod + def get_by_id(db: Session, signature_id: int) -> Optional[Signature]: + """Get signature by ID""" + return db.query(Signature).filter(Signature.id == signature_id).first() + + @staticmethod + def get_by_code(db: Session, code: str) -> Optional[Signature]: + """Get signature by code""" + return db.query(Signature).filter(Signature.code == code).first() + + @staticmethod + def create(db: Session, signature_data: SignatureCreateDTO) -> Signature: + """Create a new signature""" + try: + db_signature = Signature( + **signature_data.model_dump(exclude_unset=True)) + + db.add(db_signature) + db.commit() + db.refresh(db_signature) + + return db_signature + + except IntegrityError as e: + db.rollback() + logger.error(f"IntegrityError creating signature: {str(e)}") + raise HTTPException( + status_code=400, + detail="Signature code already exists", + ) + except Exception as e: + db.rollback() + logger.error(f"Error creating signature: {str(e)}") + raise HTTPException( + status_code=500, detail="Error creating signature") + + @staticmethod + def update( + db: Session, signature_id: int, signature_data: SignatureUpdateDTO + ) -> Optional[Signature]: + """Update a signature""" + try: + db_signature = db.query(Signature).filter( + Signature.id == signature_id + ).first() + + if not db_signature: + return None + + for key, value in signature_data.model_dump(exclude_unset=True).items(): + setattr(db_signature, key, value) + + db.commit() + db.refresh(db_signature) + + return db_signature + + except IntegrityError as e: + db.rollback() + logger.error(f"IntegrityError updating signature: {str(e)}") + raise HTTPException( + status_code=400, + detail="Error updating signature", + ) + except Exception as e: + db.rollback() + logger.error(f"Error updating signature: {str(e)}") + raise HTTPException( + status_code=500, detail="Error updating signature") + + @staticmethod + def delete(db: Session, signature_id: int) -> bool: + """Delete a signature""" + try: + db_signature = db.query(Signature).filter( + Signature.id == signature_id + ).first() + + if not db_signature: + return False + + db.delete(db_signature) + db.commit() + + return True + + except Exception as e: + db.rollback() + logger.error(f"Error deleting signature: {str(e)}") + raise HTTPException( + status_code=500, detail="Error deleting signature")