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

This commit is contained in:
2025-12-06 23:39:19 -06:00
parent 642f31871f
commit ad10777810
5 changed files with 349 additions and 0 deletions

View File

@@ -0,0 +1,3 @@
"""
Módulo de firmas
"""

View File

@@ -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

View File

@@ -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"<Signature(id={self.id}, code={self.code})>"

View File

@@ -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

View File

@@ -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")