Merge remote-tracking branch 'origin/development' into feature/process-invoices

This commit is contained in:
2026-03-13 11:06:03 -05:00
41 changed files with 2840 additions and 1029 deletions

View File

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

View File

@@ -1,43 +0,0 @@
"""
DTOs (Data Transfer Objects) para módulo de localización
"""
from typing import Optional
from pydantic import BaseModel, Field
class LocationCreateDTO(BaseModel):
"""DTO para crear una localización"""
code: str = Field(..., max_length=5, description="Location code")
description: Optional[str] = Field(
None, max_length=200, description="Location description"
)
class Config:
from_attributes = True
class LocationUpdateDTO(BaseModel):
"""DTO para actualizar una localización"""
code: Optional[str] = Field(
None, max_length=5, description="Location code")
description: Optional[str] = Field(
None, max_length=200, description="Location description"
)
class Config:
from_attributes = True
class LocationResponseDTO(BaseModel):
"""DTO para responder con datos de una localización"""
id: int
code: str
description: Optional[str] = None
class Config:
from_attributes = True

View File

@@ -1,36 +0,0 @@
"""
Modelos ORM para gestión de localización
"""
from typing import Optional
from api.v1.common.base_models import TenantScopedMixin, TimestampMixin
from core.database import Base
from sqlalchemy import Integer, PrimaryKeyConstraint, String, UniqueConstraint
from sqlalchemy.orm import Mapped, mapped_column
class Location(Base, TenantScopedMixin, TimestampMixin):
"""
Modelo para la tabla Location - Localización
"""
__tablename__ = "location" # SLocalizacion
__table_args__ = (
PrimaryKeyConstraint("id", name="location_pkey"),
UniqueConstraint("code", name="location_code_unique"),
{"schema": "a24"},
)
# Primary key
id: Mapped[int] = mapped_column(
Integer, primary_key=True, autoincrement=True)
# Location code (unique)
code: Mapped[str] = mapped_column(String(5), nullable=False, unique=True)
# Location description
description: Mapped[Optional[str]] = mapped_column(String(200))
def __repr__(self):
return f"<Location(id={self.id}, code={self.code}, description={self.description})>"

View File

@@ -1,136 +0,0 @@
"""
Rutas para gestión de localización
"""
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 LocationCreateDTO, LocationResponseDTO, LocationUpdateDTO
from .models import Location
from .service import LocationService
router = APIRouter(prefix="/locations", tags=["locations"])
@router.get(
"",
response_model=dict,
summary="Get all locations",
)
async def get_all_locations(
skip: int = Query(0, ge=0),
limit: int = Query(50, ge=1, le=100),
code: str = Query(None),
description: str = Query(None),
db: Session = Depends(get_core_db),
):
"""Get all locations with optional filtering and pagination"""
filters = {}
if code:
filters["code"] = code
if description:
filters["description"] = description
locations, total = LocationService.get_all(db, skip, limit, filters)
return {
"data": [LocationResponseDTO.model_validate(location) for location in locations],
"total": total,
"skip": skip,
"limit": limit,
}
@router.get(
"/{location_id}",
response_model=LocationResponseDTO,
summary="Get location by ID",
)
async def get_location(
location_id: int,
db: Session = Depends(get_core_db),
):
"""Get a location by its ID"""
location = LocationService.get_by_id(db, location_id)
if not location:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Location not found",
)
return LocationResponseDTO.model_validate(location)
@router.get(
"/code/{code}",
response_model=LocationResponseDTO,
summary="Get location by code",
)
async def get_location_by_code(
code: str,
db: Session = Depends(get_core_db),
):
"""Get a location by its code"""
location = LocationService.get_by_code(db, code)
if not location:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Location not found",
)
return LocationResponseDTO.model_validate(location)
@router.post(
"",
response_model=LocationResponseDTO,
status_code=status.HTTP_201_CREATED,
summary="Create location",
)
async def create_location(
location_data: LocationCreateDTO,
db: Session = Depends(get_core_db),
):
"""Create a new location"""
location = LocationService.create(db, location_data)
return LocationResponseDTO.model_validate(location)
@router.put(
"/{location_id}",
response_model=LocationResponseDTO,
summary="Update location",
)
async def update_location(
location_id: int,
location_data: LocationUpdateDTO,
db: Session = Depends(get_core_db),
):
"""Update a location"""
location = LocationService.update(db, location_id, location_data)
if not location:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Location not found",
)
return LocationResponseDTO.model_validate(location)
@router.delete(
"/{location_id}",
status_code=status.HTTP_204_NO_CONTENT,
summary="Delete location",
)
async def delete_location(
location_id: int,
db: Session = Depends(get_core_db),
):
"""Delete a location"""
success = LocationService.delete(db, location_id)
if not success:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Location not found",
)
return None

View File

@@ -1,136 +0,0 @@
"""
Capa de servicio para lógica de negocio de localización
"""
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 LocationCreateDTO, LocationResponseDTO, LocationUpdateDTO
from .models import Location
logger = logging.getLogger(__name__)
class LocationService:
"""Servicio para gestión de localización"""
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[Location], int]:
"""Get all locations with pagination"""
query = db.query(Location)
if filters:
if filters.get("code"):
query = query.filter(
Location.code.ilike(f"%{filters['code']}%"))
if filters.get("description"):
query = query.filter(
Location.description.ilike(f"%{filters['description']}%")
)
total = query.count()
locations = query.offset(skip).limit(limit).all()
return locations, total
@staticmethod
def get_by_id(db: Session, location_id: int) -> Optional[Location]:
"""Get location by ID"""
return db.query(Location).filter(Location.id == location_id).first()
@staticmethod
def get_by_code(db: Session, code: str) -> Optional[Location]:
"""Get location by code"""
return db.query(Location).filter(Location.code == code).first()
@staticmethod
def create(db: Session, location_data: LocationCreateDTO) -> Location:
"""Create a new location"""
try:
db_location = Location(
**location_data.model_dump(exclude_unset=True))
db.add(db_location)
db.commit()
db.refresh(db_location)
return db_location
except IntegrityError as e:
db.rollback()
logger.error(f"IntegrityError creating location: {str(e)}")
raise HTTPException(
status_code=400,
detail="Location code already exists",
)
except Exception as e:
db.rollback()
logger.error(f"Error creating location: {str(e)}")
raise HTTPException(
status_code=500, detail="Error creating location")
@staticmethod
def update(
db: Session, location_id: int, location_data: LocationUpdateDTO
) -> Optional[Location]:
"""Update a location"""
try:
db_location = db.query(Location).filter(
Location.id == location_id).first()
if not db_location:
return None
for key, value in location_data.model_dump(exclude_unset=True).items():
setattr(db_location, key, value)
db.commit()
db.refresh(db_location)
return db_location
except IntegrityError as e:
db.rollback()
logger.error(f"IntegrityError updating location: {str(e)}")
raise HTTPException(
status_code=400,
detail="Error updating location",
)
except Exception as e:
db.rollback()
logger.error(f"Error updating location: {str(e)}")
raise HTTPException(
status_code=500, detail="Error updating location")
@staticmethod
def delete(db: Session, location_id: int) -> bool:
"""Delete a location"""
try:
db_location = db.query(Location).filter(
Location.id == location_id).first()
if not db_location:
return False
db.delete(db_location)
db.commit()
return True
except Exception as e:
db.rollback()
logger.error(f"Error deleting location: {str(e)}")
raise HTTPException(
status_code=500, detail="Error deleting location")