diff --git a/backend/api/v1/modules/a24/inv/location/__init__.py b/backend/api/v1/modules/a24/inv/location/__init__.py new file mode 100644 index 00000000..935e07ae --- /dev/null +++ b/backend/api/v1/modules/a24/inv/location/__init__.py @@ -0,0 +1,3 @@ +""" +Módulo de localización +""" diff --git a/backend/api/v1/modules/a24/inv/location/dto.py b/backend/api/v1/modules/a24/inv/location/dto.py new file mode 100644 index 00000000..a6fc6735 --- /dev/null +++ b/backend/api/v1/modules/a24/inv/location/dto.py @@ -0,0 +1,43 @@ +""" +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 diff --git a/backend/api/v1/modules/a24/inv/location/models.py b/backend/api/v1/modules/a24/inv/location/models.py new file mode 100644 index 00000000..e3f39813 --- /dev/null +++ b/backend/api/v1/modules/a24/inv/location/models.py @@ -0,0 +1,36 @@ +""" +Modelos ORM para gestión de localización +""" + +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 Location(Base, TenantScopedMixin): + """ + 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"" diff --git a/backend/api/v1/modules/a24/inv/location/routes.py b/backend/api/v1/modules/a24/inv/location/routes.py new file mode 100644 index 00000000..0920bd4e --- /dev/null +++ b/backend/api/v1/modules/a24/inv/location/routes.py @@ -0,0 +1,136 @@ +""" +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 diff --git a/backend/api/v1/modules/a24/inv/location/service.py b/backend/api/v1/modules/a24/inv/location/service.py new file mode 100644 index 00000000..0ef5726c --- /dev/null +++ b/backend/api/v1/modules/a24/inv/location/service.py @@ -0,0 +1,136 @@ +""" +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")